introducing the schedule engine

Author: ericallam

apps/webapp/app/env.server.ts

@@ -683,6 +683,7 @@ const EnvironmentSchema = z.object({
   COMMON_WORKER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
   COMMON_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
 
+  SCHEDULE_ENGINE_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
   SCHEDULE_WORKER_ENABLED: z.string().default(process.env.WORKER_ENABLED ?? "true"),
   SCHEDULE_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(1),
   SCHEDULE_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(1),

apps/webapp/app/services/worker.server.ts

@@ -22,7 +22,6 @@ import { ResumeTaskDependencyService } from "~/v3/services/resumeTaskDependency.
 import { ResumeTaskRunDependenciesService } from "~/v3/services/resumeTaskRunDependencies.server";
 import { RetryAttemptService } from "~/v3/services/retryAttempt.server";
 import { TimeoutDeploymentService } from "~/v3/services/timeoutDeployment.server";
-import { TriggerScheduledTaskService } from "~/v3/services/triggerScheduledTask.server";
 import { GraphileMigrationHelperService } from "./db/graphileMigrationHelper.server";
 import { sendEmail } from "./email.server";
 import { reportInvocationUsage } from "./platform.v3.server";
@@ -32,6 +31,7 @@ import {
   BatchProcessingOptions as RunEngineBatchProcessingOptions,
   RunEngineBatchTriggerService,
 } from "~/runEngine/services/batchTrigger.server";
+import { scheduleEngine } from "~/v3/scheduleEngine.server";
 
 const workerCatalog = {
   scheduleEmail: DeliverEmailSchema,
@@ -222,13 +222,10 @@ function getWorkerQueue() {
         priority: 0,
         maxAttempts: 3, // total delay of 30 seconds
         handler: async (payload, job) => {
-          const service = new TriggerScheduledTaskService();
-
-          // NOTE: This graphile worker job is now part of a graceful migration to Redis workers.
-          // When this job runs, it will execute the current schedule and automatically
-          // enqueue the next execution in the new Redis-based schedule worker with
-          // distributed timing. Eventually this graphile job will be removed.
-          return await service.call(payload.instanceId, job.attempts === job.max_attempts);
+          await scheduleEngine.triggerScheduledTask({
+            instanceId: payload.instanceId,
+            finalAttempt: job.attempts === job.max_attempts,
+          });
         },
       },
       "v3.performTaskRunAlerts": {

apps/webapp/app/v3/scheduleEngine.server.ts

@@ -0,0 +1,122 @@
+import { ScheduleEngine, TriggerScheduledTaskCallback } from "@internal/schedule-engine";
+import { stringifyIO } from "@trigger.dev/core/v3";
+import { prisma } from "~/db.server";
+import { env } from "~/env.server";
+import { devPresence } from "~/presenters/v3/DevPresence.server";
+import { logger } from "~/services/logger.server";
+import { singleton } from "~/utils/singleton";
+import { TriggerTaskService } from "./services/triggerTask.server";
+import { meter, tracer } from "./tracer.server";
+
+export const scheduleEngine = singleton("ScheduleEngine", createScheduleEngine);
+
+export type { ScheduleEngine };
+
+const triggerScheduledTaskCallback: TriggerScheduledTaskCallback = async ({
+  taskIdentifier,
+  environment,
+  payload,
+  scheduleInstanceId,
+  scheduleId,
+  exactScheduleTime,
+}) => {
+  try {
+    // Use TriggerTaskServiceV1 for now (can be updated to use TriggerTaskService when ready)
+    const triggerService = new TriggerTaskService();
+
+    const payloadPacket = await stringifyIO(payload);
+
+    logger.debug("Triggering scheduled task", {
+      taskIdentifier,
+      environment,
+      payload,
+      scheduleInstanceId,
+      scheduleId,
+      exactScheduleTime,
+    });
+
+    const result = await triggerService.call(
+      taskIdentifier,
+      environment,
+      { payload: payloadPacket.data, options: { payloadType: payloadPacket.dataType } },
+      {
+        customIcon: "scheduled",
+        scheduleId,
+        scheduleInstanceId,
+        queueTimestamp: exactScheduleTime,
+      }
+    );
+
+    return { success: !!result };
+  } catch (error) {
+    return {
+      success: false,
+      error: error instanceof Error ? error.message : String(error),
+    };
+  }
+};
+
+async function isDevEnvironmentConnectedHandler(environmentId: string) {
+  const environment = await prisma.runtimeEnvironment.findFirst({
+    where: {
+      id: environmentId,
+    },
+    select: {
+      currentSession: {
+        select: {
+          disconnectedAt: true,
+        },
+      },
+      project: {
+        select: {
+          engine: true,
+        },
+      },
+    },
+  });
+
+  if (!environment) {
+    return false;
+  }
+
+  if (environment.project.engine === "V1") {
+    const v3Disconnected = !environment.currentSession || environment.currentSession.disconnectedAt;
+
+    return !v3Disconnected;
+  }
+
+  const v4Connected = await devPresence.isConnected(environmentId);
+
+  return v4Connected;
+}
+
+function createScheduleEngine() {
+  const engine = new ScheduleEngine({
+    prisma,
+    logLevel: env.SCHEDULE_ENGINE_LOG_LEVEL,
+    redis: {
+      host: env.SCHEDULE_WORKER_REDIS_HOST ?? "localhost",
+      port: env.SCHEDULE_WORKER_REDIS_PORT ?? 6379,
+      username: env.SCHEDULE_WORKER_REDIS_USERNAME,
+      password: env.SCHEDULE_WORKER_REDIS_PASSWORD,
+      keyPrefix: "schedule:",
+      enableAutoPipelining: true,
+      ...(env.SCHEDULE_WORKER_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+    },
+    worker: {
+      concurrency: env.SCHEDULE_WORKER_CONCURRENCY_LIMIT,
+      pollIntervalMs: env.SCHEDULE_WORKER_POLL_INTERVAL,
+      shutdownTimeoutMs: env.SCHEDULE_WORKER_SHUTDOWN_TIMEOUT_MS,
+      disabled: env.SCHEDULE_WORKER_ENABLED === "0",
+    },
+    distributionWindow: {
+      seconds: env.SCHEDULE_WORKER_DISTRIBUTION_WINDOW_SECONDS,
+    },
+    tracer,
+    meter,
+    onTriggerScheduledTask: triggerScheduledTaskCallback,
+    isDevEnvironmentConnectedHandler: isDevEnvironmentConnectedHandler,
+  });
+
+  return engine;
+}