wehub-resource-sync
122 lines
3.4 KiB
TypeScript
122 lines
3.4 KiB
TypeScript
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,
|
|
});
|
|
}
|
|
}
|