Human-in-the-Loop

import { serve } from "@upstash/workflow/nextjs";
import { realtime } from "@/lib/realtime";

type WorkflowPayload = {
  userId: string;
  action: string;
};

export const { POST } = serve<WorkflowPayload>(
  async (context) => {
    const { userId, action } = context.requestPayload;
    const workflowRunId = context.workflowRunId;
    const channel = realtime.channel(workflowRunId);

    // Emit run start
    await context.run("start-workflow", () =>
      channel.emit("workflow.update", {
        type: "runStart",
        workflowRunId,
        timestamp: Date.now(),
      })
    );

    // Step 1: Initial Processing
    await context.run("initial-processing", async () => {
      await channel.emit("workflow.update", {
        type: "stepStart",
        workflowRunId,
        stepName: "initial-processing",
        timestamp: Date.now(),
      });

      // Your processing logic
      const result = {
        preprocessed: true,
        userId,
        action,
        requiresApproval: true
      };

      await channel.emit("workflow.update", {
        type: "stepFinish",
        workflowRunId,
        stepName: "initial-processing",
        timestamp: Date.now(),
        result,
      });

      return result;
    });

    // Step 2: Wait for Human Approval
    const eventId = `approval-${workflowRunId}`;

    const [{ eventData, timeout }] = await Promise.all([
      // Wait for approval event
      context.waitForEvent<{ approved: boolean }>(
        "wait-for-approval",
        eventId,
        { timeout: "5m" }
      ),
      // Notify frontend that we're waiting
      context.run("notify-waiting", () =>
        channel.emit("workflow.update", {
          type: "waitingForInput",
          workflowRunId,
          eventId,
          message: `Waiting for approval to process action: ${action}`,
          timestamp: Date.now(),
        })
      ),
    ]);

    // Handle timeout
    if (timeout) {
      await channel.emit("workflow.update", {
        type: "stepFail",
        workflowRunId,
        stepName: "wait-for-approval",
        timestamp: Date.now(),
        error: "Approval timeout - no response received within 5 minutes",
      });
      return { success: false, reason: "timeout" };
    }

    // Notify that input was resolved
    await channel.emit("workflow.update", {
      type: "inputResolved",
      workflowRunId,
      eventId,
      timestamp: Date.now(),
    });

    const status = eventData.approved ? "approved" : "rejected"

    // Step 3: Process based on approval
    await context.run(`process-${status}`, async () => {
      await channel.emit("workflow.update", {
        type: "stepStart",
        workflowRunId,
        stepName: `process-${status}`,
        timestamp: Date.now(),
      });

      const result = {
        status,
        processedAt: Date.now(),
        action,
        userId,
      };

      await channel.emit("workflow.update", {
        type: "stepFinish",
        workflowRunId,
        stepName: `process-${status}`,
        timestamp: Date.now(),
        result,
      });

      return result;
    });

    // Step 4: Finalize (only if approved)
    if (eventData.approved) {
      // Additional steps...
    }

    // Emit completion
    await channel.emit("workflow.update", {
      type: "runFinish",
      workflowRunId,
      timestamp: Date.now(),
      status: "success",
    });

    return {
      success: true,
      approved: eventData.approved,
      workflowRunId
    };
  },
  {
    failureFunction: async ({ context }) => {
      const workflowRunId = context.workflowRunId;
      const channel = realtime.channel(workflowRunId);

      await channel.emit("workflow.update", {
        type: "runFinish",
        workflowRunId,
        timestamp: Date.now(),
        status: "failed",
        error: "Workflow execution failed",
      });
    },
  }
);

import { NextRequest, NextResponse } from "next/server";
import { workflowClient } from "@/lib/workflow";

export async function POST(request: NextRequest) {
  try {
    const body = await request.json();
    const { eventId, eventData } = body;

    if (!eventId) {
      return NextResponse.json(
        { success: false, error: "eventId is required" },
        { status: 400 }
      );
    }

    // Notify the workflow
    await workflowClient.notify({
      eventId,
      eventData,
    });

    return NextResponse.json({ success: true });
  } catch (error) {
    console.error("Error notifying workflow:", error);
    return NextResponse.json(
      { success: false, error: "Failed to notify workflow" },
      { status: 500 }
    );
  }
}

"use client";

import { useRealtime } from "@upstash/realtime/client";
import { useState, useCallback, useRef } from "react";
import type { RealtimeEvents } from "@/lib/realtime";

interface WorkflowStep {
  stepName: string;
  status: "running" | "completed" | "failed";
  timestamp: number;
  error?: string;
  result?: unknown;
}

interface WaitingState {
  eventId: string;
  message: string;
  timestamp: number;
}

export function useWorkflowWithRealtime() {
  const [workflowRunId, setWorkflowRunId] = useState<string | null>(null);
  const [steps, setSteps] = useState<WorkflowStep[]>([]);
  const [waitingState, setWaitingState] = useState<WaitingState | null>(null);
  const [isTriggering, setIsTriggering] = useState(false);
  const [runStatus, setRunStatus] = useState<{
    status: "running" | "success" | "failed";
    error?: string;
  } | null>(null);
  
  // Track resolved event IDs to ignore historical waiting states
  const resolvedEventIdsRef = useRef<Set<string>>(new Set());

  useRealtime<RealtimeEvents>({
    enabled: !!workflowRunId,
    channels: workflowRunId ? [workflowRunId] : [],
    event: "workflow.update",
    history: true,
    onData(data) {
      if (data.type === "runStart") {
        setRunStatus({ status: "running" });
      } else if (data.type === "runFinish") {
        setRunStatus({
          status: data.status,
          error: data.error,
        });
      } else if (data.type === "inputResolved") {
        // Mark this event as resolved
        resolvedEventIdsRef.current.add(data.eventId);
        // Clear waiting state if it matches
        setWaitingState((prev) => 
          prev?.eventId === data.eventId ? null : prev
        );
      } else if (data.type === "stepStart") {
        setSteps((prev) => [
          ...prev,
          {
            stepName: data.stepName,
            status: "running",
            timestamp: data.timestamp,
          },
        ]);
      } else if (data.type === "stepFinish") {
        setSteps((prev) =>
          prev.map((step) =>
            step.stepName === data.stepName
              ? { ...step, status: "completed", result: data.result }
              : step
          )
        );
      } else if (data.type === "stepFail") {
        setSteps((prev) =>
          prev.map((step) =>
            step.stepName === data.stepName
              ? { ...step, status: "failed", error: data.error }
              : step
          )
        );
      } else if (data.type === "waitingForInput") {
        // Only show waiting state if not already resolved
        if (!resolvedEventIdsRef.current.has(data.eventId)) {
          setWaitingState({
            eventId: data.eventId,
            message: data.message,
            timestamp: data.timestamp,
          });
        }
      }
    },
  });

  const trigger = useCallback(async () => {
    setIsTriggering(true);
    setSteps([]);
    setWaitingState(null);
    setRunStatus(null);
    resolvedEventIdsRef.current.clear();

    try {
      const response = await fetch("/api/trigger", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ workflowType: "human-in-loop" }),
      });

      const data = await response.json();
      setWorkflowRunId(data.workflowRunId);
    } catch (error) {
      console.error("Error triggering workflow:", error);
    } finally {
      setIsTriggering(false);
    }
  }, []);

  const continueWorkflow = useCallback(
    async (data: { approved: boolean }) => {
      if (!waitingState) {
        throw new Error("No workflow waiting for input");
      }

      try {
        const response = await fetch("/api/notify", {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({
            eventId: waitingState.eventId,
            eventData: data,
          }),
        });

        if (!response.ok) {
          throw new Error("Failed to notify workflow");
        }

        // The waiting state will be cleared when we receive inputResolved event
      } catch (error) {
        console.error("Error continuing workflow:", error);
        throw error;
      }
    },
    [waitingState]
  );

  return {
    trigger,
    continueWorkflow,
    isTriggering,
    workflowRunId,
    steps,
    waitingState,
    runStatus,
  };
}

"use client";

import { useWorkflowWithRealtime } from "@/hooks/useWorkflowWithRealtime";

export default function WorkflowPage() {
  const {
    trigger,
    isTriggering,
    steps,
    runStatus,
    waitingState,
    continueWorkflow,
  } = useWorkflowWithRealtime();

  return (
    <div
      style={{
        maxWidth: "600px",
        margin: "40px auto",
        fontFamily: "Arial, sans-serif",
      }}
    >
      <button onClick={trigger} disabled={isTriggering}>
        {isTriggering ? "Starting..." : "Click to Trigger Workflow"}
      </button>

      <h3 style={{ marginTop: "20px" }}>Run Status:</h3>

      {runStatus && <div>{runStatus.status}</div>}

      {/* Show workflow steps */}
      <h3 style={{ marginTop: "20px" }}>Workflow Steps:</h3>
      <div>
        {steps.map((step, index) => (
          <div key={index}>
            <strong>{step.stepName}</strong>: {step.status}
            {step.error && <span> - {step.error}</span>}
          </div>
        ))}
      </div>

      {/* Show approval UI when waiting for input */}
      {waitingState && (
        <div style={{ marginTop: "20px" }} className="approval-prompt">
          <p>{waitingState.message}</p>
          <p>
            <button onClick={() => continueWorkflow({ approved: true })}>
              Click to Approve
            </button>
          </p>
          <p>
            <button onClick={() => continueWorkflow({ approved: false })}>
              Click to Reject
            </button>
          </p>
        </div>
      )}
    </div>
  );
}