triggerdotdev--trigger.dev/apps/webapp/app/routes/api.v1.runs.$runParam.replay.ts

import type { ActionFunctionArgs } from "@remix-run/server-runtime";
import { json } from "@remix-run/server-runtime";
import type { TaskRun } from "@trigger.dev/database";
import { z } from "zod";
import { prisma } from "~/db.server";
import { runStore } from "~/v3/runStore.server";
import { authenticateApiRequest } from "~/services/apiAuth.server";
import { logger } from "~/services/logger.server";
import { ReplayTaskRunService } from "~/v3/services/replayTaskRun.server";
import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
import { sanitizeTriggerSource } from "~/utils/triggerSource";
import { clientSafeErrorMessage } from "~/utils/prismaErrors";

const ParamsSchema = z.object({
  /* This is the run friendly ID */
  runParam: z.string(),
});

// Subset of TaskRun fields that ReplayTaskRunService.call actually
// reads from `existingTaskRun`. Validate the buffered fallback against
// this before casting to TaskRun so a buffer-format drift surfaces as a
// 404/422 here rather than as a silent NaN/undefined deep inside
// replay. The full TaskRun type has many more fields the service never
// touches; we only assert the ones it reads.
const BufferedReplayInputSchema = z.object({
  id: z.string(),
  friendlyId: z.string(),
  runtimeEnvironmentId: z.string(),
  taskIdentifier: z.string(),
  payload: z.string(),
  payloadType: z.string(),
  queue: z.string(),
  isTest: z.boolean(),
  traceId: z.string(),
  spanId: z.string(),
  engine: z.string(),
  runTags: z.array(z.string()),
  // Nullable / optional fields the service tolerates via `??` fallbacks.
  concurrencyKey: z.string().nullable().optional(),
  workerQueue: z.string().nullable().optional(),
  machinePreset: z.string().nullable().optional(),
  realtimeStreamsVersion: z.string().nullable().optional(),
  // ReplayTaskRunService.getExistingMetadata reads these to preserve
  // the original run's metadata on replay. Without them in the schema
  // they'd be stripped by Zod's default key-passthrough behaviour, and
  // a buffered-source replay would silently lose metadata that a
  // PG-source replay carries over.
  seedMetadata: z.string().nullable().optional(),
  seedMetadataType: z.string().nullable().optional(),
});

export async function action({ request, params }: ActionFunctionArgs) {
  // Ensure this is a POST request
  if (request.method.toUpperCase() !== "POST") {
    return { status: 405, body: "Method Not Allowed" };
  }

  // Authenticate the request
  const authenticationResult = await authenticateApiRequest(request);
  if (!authenticationResult) {
    return json({ error: "Invalid or Missing API Key" }, { status: 401 });
  }

  const parsed = ParamsSchema.safeParse(params);
  if (!parsed.success) {
    return json({ error: "Invalid or missing run ID" }, { status: 400 });
  }

  const { runParam } = parsed.data;

  try {
    const env = authenticationResult.environment;
    // PG-first. Replay works on any status per audit — no
    // filter beyond friendlyId is the existing semantic; findFirst with
    // env scoping tightens it minimally without changing behaviour for
    // a correctly-authed caller.
    let taskRun: TaskRun | null = await runStore.findRun(
      {
        friendlyId: runParam,
        runtimeEnvironmentId: env.id,
      },
      prisma
    );

    if (!taskRun) {
      // Buffered fallback. SyntheticRun carries every field
      // ReplayTaskRunService reads from a TaskRun. Validate the subset of
      // fields the service consumes (BufferedReplayInputSchema above)
      // before casting; a schema mismatch surfaces as a 404 here rather
      // than as a silent undefined deep inside the service.
      const buffered = await findRunByIdWithMollifierFallback({
        runId: runParam,
        environmentId: env.id,
        organizationId: env.organizationId,
      });
      if (buffered) {
        const parsed = BufferedReplayInputSchema.safeParse(buffered);
        if (parsed.success) {
          // Manual sync point: `BufferedReplayInputSchema` covers only
          // the subset of `TaskRun` fields `ReplayTaskRunService.call`
          // currently reads from `existingTaskRun`. The cast is `as
          // unknown as TaskRun` because the full `TaskRun` type carries
          // ~40 fields the service never touches; mirroring all of them
          // on a synthetic snapshot would be misleading. If a future
          // change to `ReplayTaskRunService` reads an additional
          // `existingTaskRun` field, **add it to the schema above** —
          // otherwise the buffered path will silently feed the service
          // `undefined` for that field while the PG-source replay
          // works. The `safeParse` + warn-log + 404 below is the
          // run-time fail-safe; this comment is the design fail-safe.
          taskRun = parsed.data as unknown as TaskRun;
        } else {
          logger.warn("replay: buffered fallback failed schema validation", {
            runParam,
            issues: parsed.error.issues.map((issue) => ({
              path: issue.path.join("."),
              code: issue.code,
            })),
          });
        }
      }
    }

    if (!taskRun) {
      return json({ error: "Run not found" }, { status: 404 });
    }

    const triggerSource = sanitizeTriggerSource(request.headers.get("x-trigger-source")) ?? "api";

    const service = new ReplayTaskRunService();
    const newRun = await service.call(taskRun, { triggerSource });

    if (!newRun) {
      return json({ error: "Failed to create new run" }, { status: 400 });
    }

    return json({
      id: newRun?.friendlyId,
    });
  } catch (error) {
    if (error instanceof Error) {
      logger.error("Failed to replay run", {
        error: {
          name: error.name,
          message: error.message,
          stack: error.stack,
        },
        run: runParam,
      });
      return json({ error: clientSafeErrorMessage(error) }, { status: 400 });
    } else {
      logger.error("Failed to replay run", { error: JSON.stringify(error), run: runParam });
      return json({ error: JSON.stringify(error) }, { status: 400 });
    }
  }
}