triggerdotdev--trigger.dev/apps/webapp/app/v3/services/enqueueDelayedRun.server.ts

import { parseNaturalLanguageDuration } from "@trigger.dev/core/v3/isomorphic";
import { logger } from "~/services/logger.server";
import { workerQueue } from "~/services/worker.server";
import { commonWorker } from "../commonWorker.server";
import { BaseService } from "./baseService.server";
import { enqueueRun } from "./enqueueRun.server";
import { ExpireEnqueuedRunService } from "./expireEnqueuedRun.server";
import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
import { isV3Disabled } from "../engineDeprecation.server";

export class EnqueueDelayedRunService extends BaseService {
  public static async enqueue(runId: string, runAt?: Date) {
    await commonWorker.enqueue({
      job: "v3.enqueueDelayedRun",
      payload: { runId },
      availableAt: runAt,
      id: `v3.enqueueDelayed:${runId}`,
    });
  }

  public static async reschedule(runId: string, runAt?: Date) {
    // We have to do this for now because it's possible that the workerQueue
    // was used when the run was first delayed, and EnqueueDelayedRunService.reschedule
    // is called from RescheduleTaskRunService, which allows the runAt to be changed
    // so if we don't dequeue the old job, we might end up with multiple jobs
    await workerQueue.dequeue(`v3.enqueueDelayedRun.${runId}`);

    await commonWorker.enqueue({
      job: "v3.enqueueDelayedRun",
      payload: { runId },
      availableAt: runAt,
      id: `v3.enqueueDelayed:${runId}`,
    });
  }

  public async call(runId: string) {
    const run = await this.runStore.findRun(
      {
        id: runId,
      },
      {
        include: {
          dependency: {
            include: {
              dependentBatchRun: {
                include: {
                  dependentTaskAttempt: {
                    include: {
                      taskRun: true,
                    },
                  },
                },
              },
              dependentAttempt: {
                include: {
                  taskRun: true,
                },
              },
            },
          },
        },
      },
      this._prisma
    );

    if (!run) {
      logger.debug("Could not find delayed run to enqueue", {
        runId,
      });

      return;
    }

    // v3 (engine V1) shutdown: don't enqueue delayed V1 runs into MarQS. v4 is unaffected.
    if (isV3Disabled() && run.engine === "V1") {
      logger.debug("[EnqueueDelayedRunService] Skipping enqueue for shut-down v3 run", { runId });
      return;
    }

    const env = await controlPlaneResolver.resolveAuthenticatedEnv(run.runtimeEnvironmentId);

    if (!env) {
      logger.debug("EnqueueDelayedRunService: environment not found", { runId });
      return;
    }

    if (run.status !== "DELAYED") {
      logger.debug("Delayed run cannot be enqueued because it's not in DELAYED status", {
        run,
      });

      return;
    }

    await this._prisma.taskRun.update({
      where: {
        id: run.id,
      },
      data: {
        status: "PENDING",
        queuedAt: new Date(),
      },
    });

    if (run.ttl) {
      const expireAt = parseNaturalLanguageDuration(run.ttl);

      if (expireAt) {
        await ExpireEnqueuedRunService.enqueue(run.id, expireAt);
      }
    }

    await enqueueRun({
      env,
      run: run,
      dependentRun:
        run.dependency?.dependentAttempt?.taskRun ??
        run.dependency?.dependentBatchRun?.dependentTaskAttempt?.taskRun,
    });
  }
}