gitfor/platform

import { WorkflowEntrypoint } from "cloudflare:workers";
import type { WorkflowEvent, WorkflowStep } from "cloudflare:workers";
import { applyRun, failRun, planRun } from "./service";
import { abandonRequeue, checkRequeue } from "./requeue";
import { MAX_CI_POLLS, MAX_ENQUEUE_ATTEMPTS } from "./requeue-policy";

/** `requeue` skips planning: an applied proposal from a queue-triggered run only needs the CI wait and re-enqueue. */
export type MergeFixParams = { runId: string; phase?: "fix" | "requeue" };

/** Plan (model calls) and apply are separate steps without retries: a retry could re-bill the models or push twice. */
export class MergeFixWorkflow extends WorkflowEntrypoint<Env, MergeFixParams> {
  override async run(event: Readonly<WorkflowEvent<MergeFixParams>>, step: WorkflowStep): Promise<void> {
    const { runId, phase = "fix" } = event.payload;
    if (phase === "requeue") {
      await requeue(runId, step);
      return;
    }
    try {
      const plan = await step.do("plan", { retries: { limit: 0, delay: "1 second" }, timeout: "10 minutes" }, () =>
        planRun(runId),
      );
      if (!plan.autoApply) return;
      const applied = await step.do("apply", { retries: { limit: 0, delay: "1 second" }, timeout: "5 minutes" }, () =>
        applyRun(runId),
      );
      if (!applied.autoApply) return;
      await requeue(runId, step);
    } catch (error) {
      await step.do("fail", () => failRun(runId, error instanceof Error ? error.message : "The fix failed"));
    }
  }
}

async function requeue(runId: string, step: WorkflowStep): Promise<void> {
  try {
    for (let poll = 1; poll <= MAX_CI_POLLS + MAX_ENQUEUE_ATTEMPTS; poll++) {
      await step.sleep(`requeue-wait-${poll}`, "2 minutes");
      const final = poll === MAX_CI_POLLS + MAX_ENQUEUE_ATTEMPTS;
      const next = await step.do(`requeue-${poll}`, { retries: { limit: 2, delay: "30 seconds" } }, () =>
        checkRequeue(runId, poll, final),