Workflows: loops, blob storage, React hooks

import { delay, doWhile, loop, map, step, times } from "veryfront/workflow";

// Repeat while condition is true
loop("refine", {
  while: (ctx) => ((ctx.review as { score?: number } | undefined)?.score ?? 0) < 0.9,
  steps: [
    step("rewrite", { agent: "writer" }),
    step("review", { agent: "reviewer" }),
  ],
});

// Execute once, then repeat until the condition is true
doWhile("poll", {
  until: (ctx) => Boolean((ctx.check as { done?: boolean } | undefined)?.done),
  steps: [
    step("check", { tool: "statusChecker" }),
    delay("wait", "5s"),
  ],
});

// Fixed iterations
times("generate", 3, [
  step("variant", { agent: "writer" }),
]);

// Map over array items
map("process", {
  items: (ctx) => (ctx.input as { urls: string[] }).urls,
  processor: step("scrape", { tool: "webScraper" }),
});

"use client";
import { useWorkflow, useWorkflowStart } from "veryfront/workflow";

export default function PipelineDashboard() {
  const { start, lastRunId } = useWorkflowStart({
    workflowId: "pipeline",
    apiBase: "/api/workflows",
  });

  return (
    <div>
      <button onClick={() => start({ topic: "AI agents" })}>
        Run Pipeline
      </button>
      {lastRunId ? <WorkflowStatus runId={lastRunId} /> : null}
    </div>
  );
}

function WorkflowStatus({ runId }: { runId: string }) {
  const { status, nodeStates } = useWorkflow({ runId });

  return (
    <div>
      <p>Status: {status}</p>
      {Object.entries(nodeStates).map(([id, state]) => <div key={id}>{id}: {state.status}</div>)}
    </div>
  );
}

// app/api/workflows/[...path]/route.ts
import { createWorkflowHandler } from "veryfront/workflow";
import { getSession } from "../../../../lib/auth.ts";
import { workflows } from "../../../../lib/workflows.ts";

export const { GET, POST } = createWorkflowHandler(workflows, {
  authorize: async (request) => (await getSession(request))?.user.id ?? null,
});

type WorkflowStatus =
  | "pending"
  | "running"
  | "waiting"
  | "completed"
  | "failed"
  | "cancelled";

type NodeStatus = "pending" | "running" | "completed" | "failed" | "skipped";

interface WorkflowNodeStateSummary {
  nodeId: string;
  status: NodeStatus;
  attempt: number;
  startedAt?: string;
  completedAt?: string;
  error?: string;
}

interface WorkflowApprovalSummary {
  id: string;
  nodeId: string;
  status: "pending";
  message: string;
  requestedAt: string;
  expiresAt?: string;
}

interface WorkflowRunSummary {
  id: string;
  workflowId: string;
  version?: string;
  status: WorkflowStatus;
  currentNodes: string[];
  nodeStates: Record<string, WorkflowNodeStateSummary>;
  pendingApprovals: WorkflowApprovalSummary[];
  createdAt: string;
  startedAt?: string;
  completedAt?: string;
  error?: { message: string; nodeId?: string };
}

// veryfront.config.ts
import { defineConfig } from "veryfront";

export default defineConfig({
  security: {
    cors: {
      origin: "https://app.example.com",
      methods: ["GET", "POST", "OPTIONS"],
      allowedHeaders: ["Authorization", "Content-Type", "X-CSRF-Token"],
      credentials: true,
    },
  },
});

// app/api/workflows/[...path]/route.ts
import { createWorkflowHandler } from "veryfront/workflow";
import { getSession } from "../../../../lib/auth.ts";
import { workflows } from "../../../../lib/workflows.ts";

const handlers = createWorkflowHandler(workflows, {
  authorize: async (request) => (await getSession(request))?.user.id ?? null,
});

export const GET = handlers.GET;
export const POST = handlers.POST;

const TERMINAL_RUN_STATUSES = new Set(["completed", "failed", "cancelled"]);

export function observeWorkflowRun(
  runId: string,
  options: { withCredentials?: boolean } = {},
): EventSource {
  const events = new EventSource(
    `/api/workflows/runs/${encodeURIComponent(runId)}/events`,
    { withCredentials: options.withCredentials ?? false },
  );

  events.addEventListener("snapshot", (message) => {
    const run = JSON.parse((message as MessageEvent).data);
    console.log(run.status, run.nodeStates);
    if (TERMINAL_RUN_STATUSES.has(run.status)) events.close();
  });

  for (
    const name of [
      "step.started",
      "step.completed",
      "step.failed",
      "step.skipped",
      "run.status",
      "approval.pending",
    ]
  ) {
    events.addEventListener(name, (message) => {
      const event = JSON.parse((message as MessageEvent).data);
      console.log(name, event);
      if (event.type === "run.status" && TERMINAL_RUN_STATUSES.has(event.status)) {
        events.close();
      }
    });
  }

  events.addEventListener("error", (event) => {
    if (event instanceof MessageEvent) {
      console.error(JSON.parse(event.data));
      events.close();
    }
  });

  return events;
}

export const { GET, POST } = createWorkflowHandler(workflows, {
  authorize: async (request) => (await getSession(request))?.user.id ?? null,
  maxEventStreams: 64,
  maxEventStreamsPerIdentity: 8,
});