diff --git a/skills/SKILL.md b/skills/SKILL.md new file mode 100644 index 00000000..11c3401f --- /dev/null +++ b/skills/SKILL.md @@ -0,0 +1,60 @@ +--- +name: @upstash/workflow TypeScript SDK Skill +description: Lightweight guidance for using the Upstash Workflow SDK to define, trigger, and manage workflows. Use this Skill whenever a user wants to create workflow endpoints, run steps, or interact with the Upstash Workflow client. +--- + +# Upstash Workflow SDK + +## Quick Start + +The Upstash Workflow SDK lets you expose serverless workflow endpoints and run them reliably using QStash under the hood. + +Install: + +```bash +npm install @upstash/workflow +``` + +Define a simple workflow endpoint: + +```ts +import { serve } from "@upstash/workflow"; + +export const { POST } = serve(async (context) => { + await context.run("step-1", () => console.log("step 1")); + await context.run("step-2", () => console.log("step 2")); +}); +``` + +Trigger it from your backend: + +```ts +import { Client } from "@upstash/workflow"; + +const client = new Client({ token: process.env.QSTASH_TOKEN! }); +await client.trigger({ url: "https://your-app.com/api/workflow" }); +``` + +## Other Skill Files + +These files contain the full documentation. Use them for details, patterns, and advanced behavior. + +- basics: + - **basics/serve** – How to expose workflow endpoints. + - **basics/context** – Full API for workflow `context` (steps, waits, webhooks, events, invoke, etc.). + - **basics/client** – Using the Workflow client to trigger, cancel, inspect, and notify runs. +- features: + - **features/invoke** – Cross‑workflow invocation. + - **features/reliability** – Retries, failure callbacks, and DLQ. + - **features/flow-control** – Rate limits, concurrency, and parallelism. + - **features/wait-for-event** – Notify and wait-for-event patterns. + - **features/webhooks** – Webhook creation and consumption. +- how to: + - **how-to/local-dev** – Local QStash dev server and tunneling. + - **how-to/realtime** – Realtime and human‑in‑the‑loop workflows. + - **how-to/migrations** – Migrating workflows safely. + - **how-to/middleware** – Adding middleware to workflows. +- other files: + - **rest-api** – Low-level REST endpoints for interacting with QStash/Workflow. + - **troubleshooting** – Common debugging and environment issues. + - **agents** – Using Workflow with agents, orchestrators, and automation patterns. diff --git a/skills/agents.md b/skills/agents.md new file mode 100644 index 00000000..3eb941c7 --- /dev/null +++ b/skills/agents.md @@ -0,0 +1,248 @@ +--- +name: workflow-agents +description: Skill for building, configuring, and orchestrating Upstash Workflow Agents. Use when the user mentions agent workflows, multi-agent collaboration, tools, tasks, Upstash Workflow, or patterns like evaluator-optimizer, prompt chaining, parallelization, or orchestrator-workers. +--- + +# Workflow Agents + +This Skill provides guidance for defining and orchestrating agents, tools, models, and tasks inside **Upstash Workflow** using **TypeScript**. It explains core concepts, common patterns, workflow configuration, and pitfalls. + +## Quick Overview + +Use this Skill when: + +- Building agents with tools and backgrounds +- Creating multi-step or multi-agent tasks +- Implementing patterns like evaluator-optimizer or prompt chaining +- Integrating LangChain or AI SDK tools +- Running workflows reliably inside Upstash Workflow + +The Workflow Agents API centers around four elements: + +- **Models** — LLM providers (OpenAI, Anthropic, or any AI SDK provider) +- **Tools** — Functions agents can call +- **Agents** — LLM instances with background, maxSteps, and tools +- **Tasks** — Executable prompts assigned to one or more agents + +--- + +# Defining Models + +A model defines which provider the agent uses and includes optional configuration for retries, timeouts, or rate‑limits. + +```ts +const agents = agentWorkflow(context); + +// Basic OpenAI model +const model = agents.openai("gpt-3.5-turbo"); + +// OpenAI-compatible provider +const deepseek = agents.openai("deepseek-chat", { + baseURL: "https://api.deepseek.com", + apiKey: process.env.DEEPSEEK_API_KEY, +}); + +// AI SDK provider (ex: Anthropic) +const anthropic = agents.AISDKModel({ + context, + provider: createAnthropic, // imported from @ai-sdk/anthropic + providerParams: { apiKey: process.env.ANTHROPIC_KEY }, + agentCallParams: { + timeout: 1000, + retries: 0, + }, +}); +``` + +**Key parameters:** + +- **callSettings / agentCallParams** — timeout, retries, flow control +- **provider / providerParams** — when using AI SDK providers + +**Pitfall:** If you use an OpenAI-compatible provider, you **must** set `baseURL`. + +--- + +# Defining Tools + +Tools extend what agents can do. Workflow supports: + +- WorkflowTool (native) +- AI SDK tools +- LangChain tools +- Agentic toolkits + +```ts +import { z } from "zod"; +import { tool } from "ai"; +import { WorkflowTool } from "@upstash/workflow-agents"; +import { WikipediaQueryRun } from "@langchain/community/tools/wikipedia_query_run"; + +const mathTool = tool({ + description: "Evaluate a math expression", + parameters: z.object({ expression: z.string() }), + execute: async ({ expression }) => mathjs.evaluate(expression), +}); + +const workflowMath = new WorkflowTool({ + description: "Evaluate math (workflow step aware)", + schema: z.object({ expression: z.string() }), + invoke: async ({ expression }) => mathjs.evaluate(expression), + executeAsStep: false, // allows context.call, etc. +}); + +const wikiTool = new WikipediaQueryRun({ + topKResults: 1, + maxDocContentLength: 500, +}); +``` + +**Common mistakes:** + +- Workflow wraps execute/invoke in `context.run` by default, so you **cannot use context.call** unless `executeAsStep: false` is set. +- LangChain tools must return **strings**, not objects. + +--- + +# Defining Agents + +Agents wrap a model and add behavior via: + +- `maxSteps` — how many LLM calls the agent is allowed to make +- `background` — system prompt +- `tools` — available actions + +```ts +const generator = agents.agent({ + model, + name: "generator", + maxSteps: 1, + background: "Generate text from prompts.", + tools: {}, +}); + +const evaluator = agents.agent({ + model, + name: "evaluator", + maxSteps: 1, + background: "Evaluate responses and give corrections.", + tools: {}, +}); +``` + +**Tips:** + +- Pick maxSteps carefully; too low prevents tool use; too high increases cost. +- Names appear in Upstash Console logs; keep them descriptive. + +--- + +# Tasks (Single & Multi Agent) + +A **task** is a single execution of an agent or a group of agents. + +```ts +// Single agent task +const single = agents.task({ + agent: generator, + prompt: "Explain quantum mechanics.", +}); +const { text } = await single.run(); + +// Multi-agent with manager agent +const multi = agents.task({ + model, // manager LLM + agents: [generator, evaluator], + maxSteps: 3, + prompt: "Generate text and refine it until quality improves.", +}); +const result = await multi.run(); +``` + +**Tip:** In multi-agent mode, the model becomes a "manager" system that decides which agent to call. + +--- + +# Common Agent Patterns + +Below are the patterns supported by the source files in this skill. + +## Prompt Chaining + +Sequential agent calls where each output becomes the next input. +Useful for: stepwise research, multi‑stage content generation, breaking down complex tasks. + +Pitfall: watch `maxSteps` for agents that need both tool calls and summarization. + +## Evaluator‑Optimizer + +Loop until evaluator returns a PASS. Simple feedback‑refinement pattern. + +Pitfall: Always check evaluator output with `.includes("PASS")`, not strict equality. + +## Parallelization + +Use multiple agents with `Promise.all` and then aggregate. + +Pitfall: Avoid extremely large aggregated prompts; summarizing before combining is recommended. + +## Orchestrator‑Workers + +Manager delegates sub‑tasks to specialized workers. +Useful for structured Q&A, multi‑topic analysis, or complex synthesis. + +Pitfall: The manager must have enough `maxSteps` to orchestrate multiple workers. + +--- + +# Best Practices + +- Give each agent a _clear_ background; ambiguous roles cause incorrect tool use. +- Define tools with strict schemas so LLMs call them reliably. +- Use multi-agent tasks when the problem requires specialization. +- Inspect console logs to debug tool calls and agent decisions. +- Use local QStash dev server during development to avoid rate limits. + +--- + +# Example: Combined Setup (models + tools + agents + tasks) + +This shows all core fields together in one concise example. + +```ts +export const { POST } = serve(async (context) => { + const agents = agentWorkflow(context); + const model = agents.openai("gpt-4o"); + + const mathTool = tool({ + description: "Compute math", + parameters: z.object({ expression: z.string() }), + execute: async ({ expression }) => mathjs.evaluate(expression), + }); + + const researcher = agents.agent({ + model, + name: "researcher", + maxSteps: 2, + background: "Research topics using wiki.", + tools: { wikiTool }, + }); + + const mathematician = agents.agent({ + model, + name: "math", + maxSteps: 2, + background: "Solve numeric problems.", + tools: { mathTool }, + }); + + const task = agents.task({ + model, // manager + agents: [researcher, mathematician], + maxSteps: 3, + prompt: "Tell me about 3 stars and compute the sum of their masses.", + }); + + return (await task.run()).text; +}); +``` diff --git a/skills/basics/client.md b/skills/basics/client.md new file mode 100644 index 00000000..4f3a039a --- /dev/null +++ b/skills/basics/client.md @@ -0,0 +1,180 @@ +# Workflow Client Basics + +This skill provides guidance for using the Workflow Client to trigger, cancel, inspect, resume, and manage workflow runs. It focuses on TypeScript usage patterns, argument structures, and common pitfalls. + +--- + +## Core Operations + +### Triggering Workflow Runs (`client.trigger`) + +Starts one or multiple workflow runs with optional payload, metadata, retry logic, and flow control. + +```ts +// Single and multiple triggers +const single = await client.trigger({ + url: "https://your-endpoint/workflow", // required + body: { msg: "hello" }, // optional payload + headers: { "x-user": "123" }, // optional headers + workflowRunId: "custom-id", // custom unique identifier + retries: 3, // retry count for steps + retryDelay: "1000 * (1 + retried)", // expression-based retry delay + delay: "10s", // start later + notBefore: Math.floor(Date.now() / 1000) + 60, // absolute schedule + label: "my-run", // dashboard filtering + disableTelemetry: true, // opt‑out telemetry + flowControl: { + // concurrency & rate limiting + key: "user-key", + rate: 10, + parallelism: 5, + period: "10m", + }, +}); + +const multiple = await client.trigger([ + { url: "https://api/runA", body: { a: 1 } }, + { url: "https://api/runB", retries: 5 }, +]); +``` + +**Pitfalls**: + +- `workflowRunId` must be unique; collisions cause errors. +- `delay` is ignored if `notBefore` is provided. +- Using `url` incorrectly (missing protocol or dynamic segments) leads to invalid workflow endpoints. + +--- + +### Canceling Workflow Runs (`client.cancel`) + +Cancel runs by ID, by URL prefix, or all active/pending runs. + +```ts +await client.cancel({ ids: ["wfr_123", "wfr_456"] }); +await client.cancel({ urlStartingWith: "https://your-endpoint.com" }); +await client.cancel({ all: true }); +``` + +- `ids` accepts both string and array, but mixing with other filters is not supported. +- `urlStartingWith` cancels _all_ descendant paths—use carefully. + +--- + +### Retrieving Logs (`client.logs`) + +Fetch workflow execution logs with filtering and cursor-based pagination. + +```ts +const { runs, cursor } = await client.logs({ + workflowRunId: "wfr_123", // filter a specific run + workflowUrl: "https://endpoint", // fetch logs for an endpoint + state: "RUN_FAILED", // filter by execution state + count: 50, // limit return size + workflowCreatedAt: 1700000000, // Unix timestamp + cursor: undefined, // pagination +}); +``` + +--- + +### Notifying Events (`client.notify`) + +Notify workflows paused at `context.waitForEvent`. + +```ts +await client.notify({ + eventId: "order-paid", + eventData: { orderId: 1 }, // delivered to waiting workflow +}); +``` + +**Pitfall**: If no workflow is waiting, no error is thrown. + +--- + +### Fetching Waiting Workflows (`client.getWaiters`) + +Retrieve active waiters waiting for a specific event. + +```ts +const waiters = await client.getWaiters({ eventId: "order-paid" }); +``` + +This is useful for debugging or coordinating external signals. + +--- + +## Dead Letter Queue (DLQ) Operations + +### Listing DLQ Messages (`client.dlq.list`) + +```ts +const { messages, cursor } = await client.dlq.list({ + cursor, + count: 20, + filter: { + fromDate: Date.now() - 86400000, + toDate: Date.now(), + url: "https://your-endpoint.com", + responseStatus: 500, + }, +}); +``` + +- Date fields use Unix **ms**, not seconds. + +--- + +### Retry Failure Callback (`client.dlq.retryFailureFunction`) + +If a workflow's `failureFunction`/`failureUrl` call failed: + +```ts +await client.dlq.retryFailureFunction({ dlqId: "dlq-123" }); +``` + +--- + +### Restarting DLQ Runs (`client.dlq.restart`) + +Restart from the **beginning** of the workflow. + +```ts +await client.dlq.restart({ + dlqId: ["dlq-1", "dlq-2"], + retries: 5, + flowControl: { + key: "restart-group", + parallelism: 10, + }, +}); +``` + +--- + +### Resuming DLQ Runs (`client.dlq.resume`) + +Continue from the **failed step**. + +```ts +await client.dlq.resume({ + dlqId: "dlq-123", + retries: 3, + flowControl: { + key: "resume-group", + rate: 5, + }, +}); +``` + +**Common confusion**: + +- `restart` = new run from step 0. +- `resume` = same run continues at the failed step. + +--- + +## General Tips for TS Consumers + +- Use flow control when bulk‑triggering or restarting large batches to avoid rate limits. diff --git a/skills/basics/context.md b/skills/basics/context.md new file mode 100644 index 00000000..50f79a80 --- /dev/null +++ b/skills/basics/context.md @@ -0,0 +1,230 @@ +# Workflow Context Basics + +This skill documents all workflow context functions available inside a TypeScript workflow. It focuses on how these APIs behave, how their parameters interact, and common pitfalls when composing them into reliable long‑running workflows. + +--- + +## context.api + +`context.api` provides typed integrations for OpenAI, Anthropic, and Resend. These calls behave like `context.call` but enforce the correct operation names and body types. + +### Key points + +- Each provider exposes a `.call(stepName, config)` function. +- `token` is always required. +- The `operation` field selects the API method. +- `body` must match the provider's type definitions. +- Returns `{ status, body }`. + +### Example (combined) + +```ts +// OpenAI + Anthropic + Resend usage +const openai = await context.api.openai.call("openai chat", { + token: process.env.OPENAI_API_KEY!, + operation: "chat.completions.create", + body: { + model: "gpt-4o", + messages: [ + { role: "system", content: "Hello" }, + { role: "user", content: "Hi" }, + ], + }, +}); + +const anthropic = await context.api.anthropic.call("anthropic message", { + token: process.env.ANTHROPIC_API_KEY!, + operation: "messages.create", + body: { + model: "claude-3-5-sonnet-20241022", + max_tokens: 1024, + messages: [{ role: "user", content: "Hello" }], + }, +}); + +// Resend email +const email = await context.api.resend.call("send email", { + token: process.env.RESEND_API_KEY!, + headers: { "content-type": "application/json" }, + body: { + from: "Acme ", + to: ["someone@example.com"], + subject: "Welcome", + html: "

Hi!

", + }, +}); +``` + +--- + +## context.call + +Performs an HTTP request with long execution windows (up to 12 hours). Always returns a response, even for non‑2xx status codes. + +### Important parameters + +- `url`: required. +- `method`: defaults to `GET`. +- `headers`: object. +- `body`: body as string. +- `retries`: number of retry attempts. +- `retryDelay`: retry delay expression as string or number (in milliseconds). +- `flowControl`: throttling config. +- `timeout`: seconds for each attempt. +- `workflow`: used only when invoking a workflow inside `serveMany`. + +### Example (showing most fields) + +```ts +const response = await context.call("sync user", { + url: "https://api.example.com/user", + method: "POST", + headers: { authorization: `Bearer ${process.env.TOKEN}` }, + body: JSON.stringify({ id: userId, name }), + retries: 3, + retryDelay: "pow(2, retried)", // exponential + timeout: 20, + flowControl: { + key: "user-sync", + rate: 5, + parallelism: 2, + }, +}); +``` + +### Pitfalls + +- Localhost requests fail unless tunneled. + +--- + +## context.cancel + +Cancels the current workflow run. + +- Workflow is marked **canceled**, not failed. +- `failureFunction` does not run. +- No DLQ entry. + +### Example + +```ts +const result = await context.run("check something", () => ...); +if (result.cancel) await context.cancel(); +``` + +--- + +## context.createWebhook + +Creates a reusable webhook URL for external systems. + +### Example + +```ts +const { webhookUrl, eventId } = await context.createWebhook("create webhook"); +// Use webhookUrl with an external service +``` + +--- + +## context.invoke + +Invokes another workflow (must be served under the same `serveMany`) and waits for its completion. + +### Key parameters + +- `workflow`: the workflow definition. +- `body`: becomes `requestPayload` of invoked workflow. +- `headers`: forwarded headers. +- `workflowRunId`: optional override. +- `retries`, `retryDelay`, `flowControl`: control invocation retry strategy. + +### Example + +```ts +const { body, isFailed, isCanceled } = await context.invoke("invoke child", { + workflow: childWorkflow, + body: { id: 123 }, + headers: { foo: "bar" }, + retries: 2, +}); +``` + +--- + +## context.notify + +Notifies waiting workflows created via `context.waitForEvent`. + +### Example + +```ts +const { notifyResponse } = await context.notify("notify step", "order-updated", { + status: "shipped", +}); +``` + +`notifyResponse` lists all workflows resumed by this event. + +--- + +## context.run + +Runs custom synchronous or async logic as a step. Values must be JSON‑serializable. + +### Example + +```ts +const user = await context.run("load user", () => getUser()); + +await context.run("process user", () => process(user)); + +// Running in parallel +const a = context.run("step A", () => foo()); +const b = context.run("step B", () => bar()); +await Promise.all([a, b]); +``` + +### Pitfalls + +- Returned class instances lose methods due to serialization. Rehydrate via `Object.assign(new Class(), data)`. + +--- + +## context.sleep + +Pauses workflow execution for a duration. String or numeric seconds. + +### Example + +```ts +await context.sleep("wait 1 day", "1d"); +``` + +--- + +## context.sleepUntil + +Pauses execution until a specific `Date`, timestamp, or parseable string. + +### Example + +```ts +const ts = new Date(Date.now() + 1000 * 60 * 60); +await context.sleepUntil("wait until", ts); +``` + +--- + +## context.waitForEvent + +Waits for an event (triggered by `context.notify`) or until timeout. + +### Example + +```ts +const { eventData, timeout } = await context.waitForEvent("wait for update", "order-updated", { + timeout: "2h", +}); +``` diff --git a/skills/basics/serve.md b/skills/basics/serve.md new file mode 100644 index 00000000..0d5d1632 --- /dev/null +++ b/skills/basics/serve.md @@ -0,0 +1,70 @@ +This documentation explains how to use `serve()` to expose workflows as HTTP endpoints. + +## Core Concept + +`serve()` turns an async route function into a workflow endpoint. When a request arrives, a workflow run is created, and the route function defines the execution logic using `context.run()`. + +### Full Example With Inline Comments + +```typescript +import { serve } from "@upstash/workflow/nextjs"; +import { z } from "zod"; +import { Client, Receiver } from "@upstash/qstash"; +import { loggingMiddleware } from "@upstash/workflow"; + +type InitialPayload = { foo: string; bar: number }; + +const payloadSchema = z.object({ foo: z.string(), bar: z.number() }); + +export const { POST } = serve( + async (context) => { + const payload = context.requestPayload; + await context.run("step-1", async () => `hello ${payload.foo}`); + }, + { + // Runs when the workflow fails after all retries + failureFunction: async ({ context, failStatus, failResponse, failHeaders, failStack }) => + console.error(failResponse), + + // Hooks for step lifecycle, run lifecycle, and debug events + middlewares: [loggingMiddleware], + + // Manually parse the initial request payload before workflow execution + initialPayloadParser: (raw) => JSON.parse(raw), + + // Automatically validate + parse the request payload using Zod + schema: payloadSchema, + + // Override the full URL used to schedule subsequent workflow steps + url: "https://myapp.com/api/workflow", + + // Override only the base portion of the inferred URL (keeps the route path) + baseUrl: "https://proxy-url.com", + + // Provide a custom QStash client + qstashClient: new Client({ token: process.env.QSTASH_TOKEN! }), + + // Provide a custom QStash Receiver for verifying that requests come from QStash + receiver: new Receiver({ + currentSigningKey: process.env.QSTASH_CURRENT_SIGNING_KEY!, + nextSigningKey: process.env.QSTASH_NEXT_SIGNING_KEY!, + }), + + // Override environment variables available inside workflow execution + env: { + QSTASH_URL: "https://qstash.upstash.io", + QSTASH_TOKEN: process.env.QSTASH_TOKEN!, + }, + + // Output detailed execution logs to stdout + verbose: true, + + // Disable anonymous telemetry for this workflow endpoint + disableTelemetry: true, + } +); +``` + +## Common Pitfalls + +- **Incorrect URL inference** behind proxies or local tunnels—set `url` or `baseUrl`. diff --git a/skills/features/flow-control.md b/skills/features/flow-control.md new file mode 100644 index 00000000..0680b902 --- /dev/null +++ b/skills/features/flow-control.md @@ -0,0 +1,145 @@ +# Flow Control + +Flow Control defines how many workflow steps may run or start within a given time, ensuring predictable load and preventing overload of internal or third‑party systems. It applies per **flow control key**, allowing steps across different workflow runs to share the same limits. + +This skill documents: + +- parallelism +- rate and period +- monitoring +- common behaviors and pitfalls + +## Core Concepts + +### Flow Control Key + +A string identifier grouping related workflow steps under shared limits. +All steps triggered with the same key share the same concurrency and rate budget. + +```ts +const client = new Client({ token: "" }); + +await client.trigger({ + url: "https://", + flowControl: { + key: "user-onboarding", // Groups steps under this key + parallelism: 5, // max concurrent steps + rate: 20, // max starts per period + period: "1m", // time window + }, +}); +``` + +## Parallelism + +Controls **how many steps may run concurrently**. +Works like a token bucket: each running step consumes one token; when it finishes, the token is released. + +- If tokens are available → step starts immediately. +- If no tokens → step enters the waitlist. +- Token handoff is **not FIFO**; later steps may start earlier. + +### Example Behavior + +- `parallelism = 3` → at most 3 steps run at the same time. +- Step 4 waits until one of the first 3 completes. + +Parallelism affects **execution concurrency**, not rate‑of‑start per time window. + +## Rate & Period + +Controls **how many steps may START within a defined time window**. + +```ts +flowControl: { + key: "billing-jobs", + rate: 2, // allow 2 starts + period: "10s", // every 10 seconds + parallelism: 1, // can be combined +} +``` + +Behavior: + +- First `rate` steps in each period start immediately. +- Additional steps wait until the next window. +- Execution duration **does not** count toward the period; only start time matters. + +This is rate limiting, not concurrency control. + +## Combining Rate + Parallelism + +Both limits apply **independently**. +A step must satisfy _both_ to run. + +Example: + +- `rate = 3 per minute` +- `parallelism = 7` + +Even if concurrency slots are free, steps may be delayed by the rate window. +Even if rate window is open, steps may be delayed by concurrency. + +This enables predictable load patterns. + +## Monitoring + +Flow control queues and waitlists are visible from: + +- Console → **FlowControl** tab (waitlist size per key) +- REST API: + - List all keys + - Get details for one key + +Useful for diagnosing bottlenecks, long queues, or misconfigured limits. + +## Important Behaviors & Pitfalls + +### Limits are stored on each step + +If you deploy new limits, previously published steps keep their original configuration. + +### No per‑step config (mostly) + +All steps inherit the workflow‑run’s flowControl block. +Exceptions: + +- `context.call` – may define a separate key to isolate API throttling. +- `context.invoke` – spawns a new workflow run with its own flowControl. + +To isolate heavy or sensitive steps, move them into a separate workflow and invoke them. + +### Waitlist order is not guaranteed + +A later step may receive a token before an earlier one. +Do not rely on ordering. + +### Queueing is automatic + +Exceeding limits never rejects steps. Everything is eventually executed. + +## Full Example + +Demonstrates defining all parameters together and how they interact. + +```ts +import { Client } from "@upstash/workflow"; + +const client = new Client({ token: "" }); + +await client.trigger({ + url: "https:///process", + flowControl: { + key: "media-processing", // Shared grouping key + parallelism: 4, // Max 4 at the same time + rate: 10, // Max 10 starts + period: "60s", // in 60 seconds + }, +}); +``` + +This configuration: + +- ensures no more than 4 processing tasks run concurrently +- ensures no more than 10 tasks start per minute +- queues everything automatically beyond those limits diff --git a/skills/features/invoke.md b/skills/features/invoke.md new file mode 100644 index 00000000..8e11a435 --- /dev/null +++ b/skills/features/invoke.md @@ -0,0 +1,97 @@ +# Workflow Invocation + +This documentation covers how to invoke one workflow from another using `context.invoke`, along with how to define and expose multiple workflows using `createWorkflow` and `serveMany`. + +--- + +## Invoking Another Workflow + +A workflow can start another workflow run and wait for it to finish before continuing. This enables orchestration and dependency chains across workflows. + +Use `context.invoke` to trigger another workflow: + +```ts +const result = await context.invoke("analyze content", { + workflow: analyzeContent, // workflow object or function + body: "input payload", // request body passed to invoked workflow + headers: { "x-user": "123" }, // optional headers + retries: 5, // optional, default 3 + flowControl: { + /* ... */ + }, + workflowRunId: "optional-custom-id", +}); + +const { + body, // response returned by invoked workflow + isFailed, // true if invoked workflow failed + isCanceled, // true if invoked workflow was canceled +} = result; +``` + +### Returning a Value + +A workflow may return any serializable value, which becomes `body` for the caller. + +### Depth Limitation + +Workflow invocation depth is limited to 100. Any recursive or cyclic chain exceeding this will fail. + +--- + +## Defining Workflows Without Exposing Endpoints + +Use `createWorkflow` to define workflows as objects. These are not automatically exposed as HTTP endpoints. + +```ts +const analyzeContent = createWorkflow(async (context) => { + await context.sleep("delay", 1); + return { message: "processed" }; +}); + +const processUser = createWorkflow(async (context) => { + const { body } = await context.invoke("invoke analyze", { + workflow: analyzeContent, // type‑safe invocation + body: "user-1", + }); + + return { result: body }; +}); +``` + +--- + +## Exposing Multiple Workflows With `serveMany` + +To allow workflows to invoke each other without requiring explicit URLs, define all related workflows under a shared parent path. + +```ts +// app/workflows/[...any]/route.ts +export const { POST } = serveMany({ + analyze: analyzeContent, + process: processUser, +}); +``` + +- Keys become route names. +- All workflows that need to invoke each other must appear in the same `serveMany` call. +- Requests are routed based on the key names. + +### Triggering a Workflow + +```ts +import { Client } from "@upstash/workflow"; + +const client = new Client({ token: "" }); + +await client.trigger({ + url: "https://your-app/workflows/analyze", +}); +``` + +--- + +## Common Pitfalls + +- **Missing workflow in `serveMany`**: If a workflow invokes another that was not included in the same `serveMany` definition, the call will fail. +- **Infinite invocation loops**: Depth limit of 100 prevents runaway recursive chains. diff --git a/skills/features/retries-failures-reliability.md b/skills/features/retries-failures-reliability.md new file mode 100644 index 00000000..01adb594 --- /dev/null +++ b/skills/features/retries-failures-reliability.md @@ -0,0 +1,239 @@ +# Workflow Reliability Features + +This documentation provides a consolidated reference for reliability mechanisms in Upstash Workflow, focusing on retries, failure handling, and the Dead Letter Queue (DLQ). It is optimized for AI‑driven execution but remains clear for developers. + +--- + +## Core Reliability Concepts + +Upstash Workflow ensures that workflow runs are durable, recoverable, and traceable. Reliability is achieved through: + +- Automatic retries for transient failures +- Failure handling via `failureFunction` or `failureUrl` +- DLQ for storing permanently failed runs +- Manual recovery actions: restart, resume, rerun failure function +- Mechanisms to prevent retries when failure is intentional + +These mechanisms work together to guarantee that no failed execution is lost and that failures can be observed and handled. + +--- + +## Automatic Retries + +Workflow steps automatically retry when they fail, using configurable retry count and delay. + +### Retry Parameters + +- `retries`: How many times the step will retry +- `retryDelay`: A math expression (string) returning delay in milliseconds; can use `retried` variable + +### Common Pitfalls + +- The delay expression must be a _string_; raw numbers disable dynamic calculation. +- A workflow only moves to DLQ after **all** retries fail. + +### Example: Configure Retry Count + Delay + +```ts +import { Client, WorkflowNonRetryableError, serve } from "@upstash/workflow"; + +// Trigger with custom retry settings +const client = new Client({ token: process.env.WORKFLOW_TOKEN! }); +await client.trigger({ + url: "https://example.com/workflow", + retries: 5, + retryDelay: "(1 + retried) * 2000", // 2s, 4s, 6s... +}); + +// Workflow with steps that may intentionally skip retries +export const { POST } = serve(async (context) => { + const shouldFail = await context.run("check", () => true); + + if (shouldFail) { + // Skips retry cycle and sends directly to failure handling + throw new WorkflowNonRetryableError("Bad state"); + } + + await context.run("process", () => doWork()); +}); +``` + +--- + +## Preventing Retries + +### When to Stop Retries + +Use these when failure is expected or should halt execution immediately: + +- `WorkflowNonRetryableError` → Fail fast, trigger failure function, enter DLQ +- `context.cancel()` → Stop workflow intentionally, no DLQ, no failure function +- Conditional returns → Graceful early exit without error + +### Pitfalls + +- `context.cancel()` marks the run as **canceled**, not failed — no failure handling executes. +- Throwing any error _other than_ `WorkflowNonRetryableError` will still trigger retries. + +--- + +## Failure Handling + +You can handle workflow failures using either: + +- `failureFunction` (runs in your workflow environment) +- `failureUrl` (external callback endpoint) + +Only **one** can be configured — they are mutually exclusive. + +### Failure Function Behavior + +- Runs after all retries fail +- Receives failure metadata and workflow context +- Cannot create new workflow steps; `context` is read‑only +- Is retried if it fails +- Can be manually retried from the DLQ + +### Failure Function Parameters + +- `context.workflowRunId` +- `context.url` +- `context.requestPayload` +- `context.headers` +- `context.env` +- `failStatus` +- `failResponse` +- `failHeaders` + +### Example: Workflow + Failure Function + Prevent‑Retry Logic + +```ts +export const { POST } = serve( + async (context) => { + const ok = await context.run("verify", () => verifySomething()); + + if (!ok) { + throw new WorkflowNonRetryableError("Verification failed"); + } + + const shouldAbort = await context.run("check", () => false); + if (shouldAbort) { + await context.cancel(); + return; + } + + await context.run("execute", () => doWork()); + }, + { + failureFunction: async ({ context, failStatus, failResponse }) => { + // Logging / cleanup / alerts + await logFailure(context.workflowRunId, failStatus, failResponse); + }, + } +); +``` + +--- + +## Dead Letter Queue (DLQ) + +A workflow enters the DLQ when: + +- A step fails _after all retries_ +- A `WorkflowNonRetryableError` is thrown +- `failureFunction` also fails (visible as failed in DLQ) + +### Retention + +- Free: 3 days +- Pay‑as‑you‑go: 1 week +- Fixed: Up to 3 months + +After retention expires, items cannot be recovered. + +--- + +## DLQ Recovery Actions + +You may perform these actions from the dashboard or programmatically. + +### 1. Resume + +Continue from the failed step while keeping previous step results. + +### 2. Restart + +Start the entire workflow from scratch, discarding previous results. + +### 3. Rerun Failure Function + +Retry the failure handler **only if it previously failed**. + +### Example: All DLQ Actions + +```ts +import { Client } from "@upstash/workflow"; +const client = new Client({ token: process.env.WORKFLOW_TOKEN! }); + +await client.dlq.resume({ dlqId: "dlq-1", retries: 3 }); // Continue run +await client.dlq.restart({ dlqId: "dlq-2", retries: 3 }); // Restart +await client.dlq.retryFailureFunction({ dlqId: "dlq-3" }); // Retry failure handler +``` + +### Pitfalls + +- `retryFailureFunction` is only allowed when the failure function failed. +- `resume` cannot be used if workflow code changed before the failure point. +- `restart` re-executes everything, including expensive steps. + +--- + +## Failure URL (Advanced Option) + +Use when failure must be handled by a separate service. + +### Notes + +- Use **either** `failureFunction` **or** `failureUrl`. +- Recommended only when your primary app might be offline. +- Failure URL receives a structured JSON payload about the failed request. + +### Example + +```ts +export const { POST } = serve( + async (context) => { + await context.run("work", () => doWork()); + }, + { + failureUrl: "https://example.com/failure-handler", + } +); +``` + +--- + +## Debugging Failed Runs + +DLQ entries store: + +- Request + response headers +- Payloads +- Failure metadata +- Failure function state + +Use the dashboard filters (workflow run ID, URL, failure function state) to identify and debug root causes. + +--- + +## Summary + +This skill unifies all Workflow reliability controls: + +- Retries (automatic, configurable) +- Failure handling (`failureFunction`, `failureUrl`) +- DLQ retention + recovery +- Fast‑fail mechanisms (`WorkflowNonRetryableError`, `cancel`, conditional exits) +- Manual recovery operations (resume, restart, retry failure function) + +Together these features ensure workflows are resilient, debuggable, and recoverable under all failure scenarios. diff --git a/skills/features/wait-for-event.md b/skills/features/wait-for-event.md new file mode 100644 index 00000000..1cb03ee6 --- /dev/null +++ b/skills/features/wait-for-event.md @@ -0,0 +1,97 @@ +# Wait for Event + +Use this feature to pause workflow execution until an external event arrives. It allows building asynchronous, event‑driven workflows without holding compute resources. + +--- + +## How It Works + +`context.waitForEvent()` suspends the workflow and registers a waiter under an event ID. When a matching `notify` call is sent, the workflow resumes with the provided event data. + +• Each waiter has a timeout. If the event does not arrive in time, the workflow continues with `{ timeout: true }`. +• Maximum timeout depends on your pricing tier. +• Multiple workflow runs may wait on the same event ID; a notify call resumes all of them. + +--- + +## Key Concepts & Pitfalls + +### Event IDs + +Use unique event IDs to avoid heavy fan‑out notifications. + +Example patterns: +• `order-123-processing-complete` +• `user-42-email-verified` + +### Race Conditions + +A notify call sent _before_ the workflow begins waiting is lost. + +To avoid this: +• Always inspect the notify response. +• Retry if no waiters were found. + +--- + +## Combined Example + +Below is a single TypeScript example showing waiting for an event, handling timeouts, and safely notifying: + +```ts +import { serve } from "@upstash/workflow/nextjs"; +import { Client } from "@upstash/workflow"; + +// Workflow that waits for an event +export const { POST } = serve(async (context) => { + const { orderId } = context.requestPayload; + const eventId = `order-${orderId}-processed`; + + // Wait for the event with timeout + const { eventData, timeout } = await context.waitForEvent( + "wait-for-order-processing", // step name + eventId, // event id + { + timeout: "1d", // optional timeout + } + ); + + if (timeout) { + // handle timeout here + await context.run("timeout-handler", async () => { + console.log("Order processing timed out"); + }); + return; + } + + // Continue workflow using eventData + await context.run("process-completed-order", async () => { + console.log("Order processed:", eventData); + }); +}); + +// External notifier (safe retry pattern) +const client = new Client({ token: "" }); + +async function notifyProcessingComplete(orderId: string, payload: any) { + const eventId = `order-${orderId}-processed`; + + // First attempt - returns array of NotifyResponse + const waiters = await client.notify({ eventId, eventData: payload }); + + if (waiters > 0) return waiters; + + // Retry if no workflows were waiting + await new Promise((r) => setTimeout(r, 3000)); + return await client.notify({ eventId, eventData: payload }); +} +``` + +--- + +## Best Practices + +• Generate event IDs that uniquely identify a specific workflow instance. +• Always handle the timeout case explicitly. +• On notification, check the returned array length to detect if any workflows were waiting. +• Prefer one event ID per workflow run to avoid notifying large groups. diff --git a/skills/features/webhooks.md b/skills/features/webhooks.md new file mode 100644 index 00000000..26c614e3 --- /dev/null +++ b/skills/features/webhooks.md @@ -0,0 +1,79 @@ +## Webhooks + +This skill teaches agents how to use webhooks inside Upstash Workflow to pause execution, wait for external callbacks, and resume reliably. It covers creation, waiting, validation, and multi-step usage. + +--- + +### Key Concepts + +• **Webhook URLs are unique per context.createWebhook() call**. +• **Workflow execution pauses** when calling `context.waitForWebhook()` until the webhook receives a request or times out. +• **Lookback protection** ensures the workflow still receives early webhook calls. +• **Workflows remain dormant** during the waiting period, avoiding compute cost. +• **Webhooks are ideal for third‑party APIs** that support callback URLs. + +--- + +### Core Workflow: Create → Call External API → Wait → Resume + +Below is a consolidated example showing: +• Creating a webhook +• Passing it to an external API +• Handling the callback +• Waiting for multiple updates +• Error/timeout handling + +```ts +import { serve } from "@upstash/workflow/nextjs"; + +export const { POST } = serve(async (context) => { + // Create a unique webhook for this workflow + const webhook = await context.createWebhook("create webhook"); + + // Trigger an external service, sending it the callback URL + await context.call("trigger-external", { + url: "https://api.example.com/process", + method: "POST", + body: { webhookUrl: webhook.webhookUrl }, + }); + + // Loop: wait for progress updates until service signals completion + let updates = []; + while (true) { + const res = await context.waitForWebhook(`wait update`, webhook, "5m"); + + if (res.timeout) break; // no progress received + + const req = res.request; + const payload = await req.json(); + updates.push(payload); + + if (req.headers.get("x-task-finished") === "true") break; + } + + return updates; +}); +``` + +--- + +### Webhook Reliability: Lookback Protection + +**Common pitfall:** external systems may call the webhook _before_ the workflow reaches `waitForWebhook()`. + +Upstash automatically stores early webhook calls. The next `waitForWebhook()` will immediately resolve with the stored request. + +This avoids race conditions and eliminates the need for manual buffering. + +--- + +### When to Use Webhooks vs. Wait for Event + +Use **webhooks** when: +• integrating with third‑party APIs +• the external system invokes a callback URL +• you need race‑condition‑safe behavior + +Use **Wait for Event** when: +• you control the notifying system and can use the Workflow Client +• you don't need lookback semantics diff --git a/skills/how-to/local-dev.md b/skills/how-to/local-dev.md new file mode 100644 index 00000000..69c97a87 --- /dev/null +++ b/skills/how-to/local-dev.md @@ -0,0 +1,128 @@ +# Local Development for Upstash Workflow + +This documentation explains how to run Upstash Workflow locally using the QStash development server and how to expose your local app using a public tunnel such as ngrok. It focuses on practical usage and TypeScript integration. + +## Overview + +Upstash Workflow uses Upstash QStash under the hood. During development you can: + +- Run a **local QStash development server** to simulate workflow behavior entirely offline. +- Optionally expose your local server using **ngrok** if you want to test with the production QStash. + +Both approaches require updating environment variables and ensuring your workflow trigger URLs correctly reference your local or tunnel address. + +--- + +## 1. Start the QStash Local Development Server + +Use the QStash CLI: + +``` +npx @upstash/qstash-cli dev +``` + +The CLI prints: + +- QSTASH_TOKEN +- QSTASH_CURRENT_SIGNING_KEY +- QSTASH_NEXT_SIGNING_KEY +- Local server URL (default: `http://127.0.0.1:8080`) + +Set these values in your `.env` file so your workflow client uses the local environment. + +--- + +## 2. Set Local Environment Variables + +Use the environment values printed by the CLI: + +``` +QSTASH_URL="http://127.0.0.1:8080" +QSTASH_TOKEN="" +QSTASH_CURRENT_SIGNING_KEY="" +QSTASH_NEXT_SIGNING_KEY="" +``` + +These ensure all workflow requests are routed locally. + +--- + +## 3. Trigger Workflows Using Local URLs + +A common pattern is determining the base URL dynamically based on environment variables. Below is an example that demonstrates: + +- Local development (localhost) +- Production deployments (auto-detected env) + +```ts +import { Client } from "@upstash/workflow"; + +const client = Client(); + +// In production, Vercel sets VERCEL_URL. Otherwise use localhost. +const BASE_URL = process.env.VERCEL_URL + ? `https://${process.env.VERCEL_URL}` + : `http://localhost:3000`; + +// Trigger a workflow with retries +const { workflowRunId } = await client.trigger({ + url: `${BASE_URL}/api/workflow`, // Local or production + retries: 3, // Optional retry logic +}); + +console.log("Workflow run:", workflowRunId); +``` + +**Common mistakes:** + +- Forgetting to include the full URL including `http://` or `https://`. +- Using a production URL while the local QStash server is running. +- Missing environment variables. + +--- + +## 4. Using ngrok (Optional) + +If your workflow must be reachable from the managed Upstash servers (not local), expose your local server publicly. + +### Install & authenticate + +``` +ngrok config add-authtoken +``` + +### Start a tunnel + +``` +ngrok http 3000 +``` + +ngrok outputs a public URL: + +``` +Forwarding https://1234abcd.ngrok.io -> http://localhost:3000 +``` + +Use this public URL instead of localhost: + +```ts +const BASE_URL = "https://1234abcd.ngrok.io"; // Public tunnel + +await client.trigger({ + url: `${BASE_URL}/api/workflow`, + retries: 3, +}); +``` + +**Pitfall:** Your ngrok port must match your dev server port, otherwise all workflow calls will fail. + +--- + +## Summary + +- Start QStash locally using `qstash-cli dev`. +- Copy generated environment variables into `.env`. +- Use local URLs in TypeScript clients while developing. +- Optionally expose your server with ngrok if you need remote access. + +This setup ensures fast workflow iteration without deploying your app. diff --git a/skills/how-to/middleware.md b/skills/how-to/middleware.md new file mode 100644 index 00000000..3be475ae --- /dev/null +++ b/skills/how-to/middleware.md @@ -0,0 +1,166 @@ +# Middlewares + +This guide explains how to intercept workflow lifecycle and debug events using middleware. It covers how to attach middlewares, define event handlers, initialize resources, and combine functionality into efficient implementations. + +## Overview + +Middlewares plug into the workflow engine and listen to: + +- Lifecycle events such as run start, run completion, and step execution +- Debug messages such as errors, warnings, and informational logs + +They are useful for logging, monitoring, instrumentation, and integrating with external systems. + +## Using Built‑in Middleware + +Add middleware inside the `middlewares` array when serving a workflow: + +```ts +import { serve } from "@upstash/workflow/nextjs"; +import { loggingMiddleware } from "@upstash/workflow"; + +export const { POST } = serve( + async (ctx) => { + await ctx.run("step-1", () => "Hello World"); + }, + { + middlewares: [loggingMiddleware], // emits run/step logs + } +); +``` + +## Creating Custom Middleware + +You define middleware by instantiating `WorkflowMiddleware` with callbacks for the events you want to handle. + +### Direct Callback Definitions + +All lifecycle and debug events can be defined in one object. + +```ts +import { WorkflowMiddleware } from "@upstash/workflow"; + +const customMiddleware = new WorkflowMiddleware({ + name: "custom-logger", + callbacks: { + // Lifecycle events + runStarted: async ({ context }) => { + console.log(`Run ${context.workflowRunId} started`); + }, + beforeExecution: async ({ context, stepName }) => { + console.log(`→ Executing step: ${stepName}`); + }, + afterExecution: async ({ context, stepName, result }) => { + console.log(`✓ Step ${stepName} completed`, result); + }, + runCompleted: async ({ context, result }) => { + console.log(`Run ${context.workflowRunId} finished`, result); + }, + + // Debug events + onError: async ({ workflowRunId, error }) => { + console.error(`Error in ${workflowRunId}`, error); + }, + onWarning: async ({ workflowRunId, warning }) => { + console.warn(`Warning from ${workflowRunId}`, warning); + }, + onInfo: async ({ workflowRunId, info }) => { + console.info(`Info from ${workflowRunId}`, info); + }, + }, +}); +``` + +### Initialization With `init` + +Use `init` when you need setup logic (connecting to a DB, creating clients, loading configuration). `init` returns an object containing the callbacks. + +```ts +import { WorkflowMiddleware } from "@upstash/workflow"; + +const databaseMiddleware = new WorkflowMiddleware({ + name: "database-logger", + init: async () => { + const db = await connectToDatabase(); + + return { + runStarted: async ({ context }) => { + await db.insert({ workflowRunId: context.workflowRunId, status: "started" }); + }, + runCompleted: async ({ context, result }) => { + await db.update({ workflowRunId: context.workflowRunId, status: "completed", result }); + }, + onError: async ({ workflowRunId, error }) => { + await db.insert({ workflowRunId, level: "error", message: error.message }); + }, + }; + }, +}); +``` + +## Event Reference + +Below is a concise list of supported event handlers and what they receive. + +### Lifecycle + +- **runStarted**: `{ context }` +- **beforeExecution**: `{ context, stepName }` +- **afterExecution**: `{ context, stepName, result }` +- **runCompleted**: `{ context, result }` + +### Debug + +- **onError**: `{ workflowRunId?, error }` +- **onWarning**: `{ workflowRunId?, warning }` +- **onInfo**: `{ workflowRunId?, info }` + +## Combined Example + +Single snippet showing typical usage patterns. + +```ts +import { serve } from "@upstash/workflow/nextjs"; +import { WorkflowMiddleware } from "@upstash/workflow"; + +// Error tracking +const errorTrackingMiddleware = new WorkflowMiddleware({ + name: "errors", + callbacks: { + onError: async ({ workflowRunId, error }) => { + await fetch("https://monitoring/errors", { + method: "POST", + body: JSON.stringify({ workflowRunId, error: error.message }), + }); + }, + }, +}); + +// Performance measurement +const timings = new Map(); +const performanceMiddleware = new WorkflowMiddleware({ + name: "perf", + callbacks: { + beforeExecution: ({ stepName }) => timings.set(stepName, Date.now()), + afterExecution: ({ stepName }) => { + const t = timings.get(stepName); + if (t) console.log(`Step ${stepName} took ${Date.now() - t}ms`); + timings.delete(stepName); + }, + }, +}); + +export const { POST } = serve( + async (ctx) => { + await ctx.run("task", () => "done"); + }, + { + middlewares: [errorTrackingMiddleware, performanceMiddleware], + } +); +``` + +## Pitfalls and Tips + +- Ensure `init` returns all callbacks; missing keys simply mean the event is ignored. +- Debug events may be called without a `workflowRunId` if errors occur before run id is identified. diff --git a/skills/how-to/migrations.md b/skills/how-to/migrations.md new file mode 100644 index 00000000..b5c19f39 --- /dev/null +++ b/skills/how-to/migrations.md @@ -0,0 +1,201 @@ +# Migration Guide for Upstash Workflow (TypeScript) + +This document provides a clear, task-focused overview of migrations between Workflow versions. It highlights breaking changes, how to update your code, and common pitfalls to avoid. Code samples combine multiple parameters in each example to reduce redundancy. + +## January 2026 – Major Release 1.0.0 + +### Agents API → moved to a separate package + +The Agents API was removed from the core workflow package and placed into `@upstash/workflow-agents` so the base SDK no longer depends on large AI-related libraries. + +**Key changes**: + +- Must install and import `agentWorkflow` from the new package. +- Agents access now happens through an `agents` helper created per workflow invocation. + +```ts +// Before +import { serve } from "@upstash/workflow/nextjs"; +export const { POST } = serve(async (context) => { + const model = context.agents.openai("gpt-3.5-turbo"); + const agent = context.agents.agent({ + /* ... */ + }); + const task = context.agents.task({ + /* ... */ + }); +}); + +// After +import { serve } from "@upstash/workflow/nextjs"; +import { agentWorkflow } from "@upstash/workflow-agents"; +export const { POST } = serve(async (context) => { + const agents = agentWorkflow(context); + const model = agents.openai("gpt-3.5-turbo"); + const agent = agents.agent({ + /* ... */ + }); + const task = agents.task({ + /* ... */ + }); +}); +``` + +--- + +### `keepTriggerConfig` and `useFailureFunction` removed + +These fields were redundant—both behaviors are now always enabled. + +```ts +// Before +await client.trigger({ + url: "...", + retries: 3, + keepTriggerConfig: true, + useFailureFunction: true, +}); + +// After +await client.trigger({ url: "...", retries: 3 }); +``` + +--- + +### Configuration moved from `serve()` → `client.trigger()` + +Options such as `retries`, `flowControl`, `retryDelay`, and `failureUrl` no longer belong in `serve()`. + +```ts +// Before +export const { POST } = serve( + async (context) => { + /* ... */ + }, + { + retries: 3, + retryDelay: "1000 * (1 + retried)", + flowControl: { key: "my-key", rate: 10 }, + } +); +await client.trigger({ url: "..." }); + +// After +export const { POST } = serve(async (context) => { + /* ... */ +}); +await client.trigger({ + url: "...", + retries: 3, + retryDelay: "1000 * (1 + retried)", + flowControl: { key: "my-key", rate: 10 }, +}); +``` + +--- + +### `stringifyBody` removed from `context.call` and `context.invoke` + +Bodies must now be strings. + +```ts +// Before +await context.call("step", { + url: "https://api.example.com", + method: "POST", + body: { key: "value" }, + stringifyBody: true, +}); + +// After +await context.call("step", { + url: "https://api.example.com", + method: "POST", + body: JSON.stringify({ key: "value" }), +}); + +// Same change applies to invoke +await context.invoke("other", { + workflow: otherWorkflow, + body: JSON.stringify({ key: "value" }), +}); +``` + +--- + +### Logger removed → middleware system added + +Removed `WorkflowLogger` class and added `WorkflowMiddleware`. Updated `verbose` param to only allow `true` (in this case, `loggingMiddleware` will be used which prints to console). + +```ts +import { loggingMiddleware, WorkflowMiddleware } from "@upstash/workflow"; + +const stepFinish = new WorkflowMiddleware({ + name: "step-finish", + callbacks: { + afterExecution: async ({ stepName, result }) => + console.log(`Step ${stepName} finished`, result), + }, +}); + +export const { POST } = serve( + async (context) => { + /* ... */ + }, + { + middlewares: [loggingMiddleware, stepFinish], + } +); +``` + +--- + +## October 2024 – Migration from `@upstash/qstash` + +### Install the new package + +Replace all workflow usage from `@upstash/qstash` with `@upstash/workflow`. + +### Serve import and return shape changed + +Most environments now require destructuring `{ POST }`. + +```ts +// Before +import { serve } from "@upstash/qstash/nextjs"; +export const POST = serve(...); + +// After +import { serve } from "@upstash/workflow/nextjs"; +export const { POST } = serve(...); +``` + +### `context.call` updated + +Now returns `{ status, headers, body }` and **does not fail** the workflow if the HTTP call fails. + +```ts +const { status, headers, body } = await context.call("call-step", { + url: "https://example.com/api", + method: "POST", +}); +``` + +**Pitfall**: Old runs may not contain `status` or `headers` until fully migrated. + +### Errors renamed + +- `QStashWorkflowError` → `WorkflowError` +- `QStashWorkflowAbort` → `WorkflowAbort` + +--- + +## Summary of Common Mistakes + +- Using `context.agents` instead of the new `agentWorkflow()` helper. +- Leaving old serve-config options (`retries`, `flowControl`, etc.). +- Forgetting to `JSON.stringify` bodies. +- Assuming `context.call` failures stop workflow execution. +- Mix-and-matching `@upstash/qstash` and `@upstash/workflow` imports. + +This guide ensures smooth migration with minimal friction while highlighting where behavioral changes may break existing workflows. diff --git a/skills/how-to/realtime.md b/skills/how-to/realtime.md new file mode 100644 index 00000000..69f90787 --- /dev/null +++ b/skills/how-to/realtime.md @@ -0,0 +1,186 @@ +# Realtime Workflow Integration (TypeScript) + +This Skill provides guidance for building real‑time and human‑in‑the‑loop workflows using Upstash Workflow + Upstash Realtime. It covers event schemas, workflow patterns, emitting/receiving updates, and handling interactive pauses. + +Key capabilities: + +- Emit strongly typed workflow events. +- Subscribe to live updates in the frontend. +- Implement pause/resume workflows using `waitForEvent` and `notify`. + +--- + +## Core Concepts + +### Realtime schema design + +Define all workflow event types in one schema. Keep event definitions minimal and stable. + +```ts +// lib/realtime.ts +const schema = { + workflow: { + // Completion events + runFinish: z.object({}), + stepFinish: z.object({ stepName: z.string(), result: z.unknown().optional() }), + + // Human‑in‑the‑loop events + waitingForInput: z.object({ eventId: z.string(), message: z.string() }), + inputResolved: z.object({ eventId: z.string() }), + }, +}; + +export const realtime = new Realtime({ schema, redis }); +export type RealtimeEvents = InferRealtimeEvents; +``` + +Common mistakes: + +- Forgetting to wrap events under a namespace (e.g. `workflow.*`). + +--- + +## Emitting Events Inside Workflows + +Emit inside `context.run()` whenever possible to avoid duplicate emissions on retries. + +```ts +// app/api/workflow/route.ts +export const { POST } = serve(async (context) => { + const channel = realtime.channel(context.workflowRunId); + + // Emit step updates + await context.run("validate", async () => { + const result = { ok: true }; + await channel.emit("workflow.stepFinish", { stepName: "validate", result }); + return result; + }); + + // Final event + await context.run("finish", () => channel.emit("workflow.runFinish", {})); +}); +``` + +Pitfall: + +- Emitting events _outside_ of `context.run()` risks duplicate delivery if retries happen. + +--- + +## Human‑in‑the‑Loop Workflows + +Use `waitForEvent()` with unique `eventId` values and emit a `waitingForInput` event immediately. + +```ts +// Step that pauses for human input +const eventId = `approval-${context.workflowRunId}`; + +const [{ eventData, timeout }] = await Promise.all([ + context.waitForEvent("wait-for-approval", eventId, { timeout: "5m" }), + context.run("notify-wait", () => + channel.emit("workflow.waitingForInput", { + eventId, + message: "Waiting for approval...", + }) + ), +]); + +if (timeout) return { success: false, reason: "timeout" }; + +await context.run("resolved", () => channel.emit("workflow.inputResolved", { eventId })); +``` + +Important details: + +- Always handle timeout: otherwise the workflow may appear stuck. +- Emit `inputResolved` so the UI can clear any pending state. +- Use deterministic `eventId` prefixes per action/approval. + +--- + +## Notify Endpoint (User Response) + +The frontend sends input back to the workflow using `client.notify()`. + +```ts +// app/api/notify/route.ts +export async function POST(req: NextRequest) { + const { eventId, eventData } = await req.json(); + + await workflowClient.notify({ eventId, eventData }); // resumes workflow + return NextResponse.json({ success: true }); +} +``` + +Common mistakes: + +- Sending the wrong `eventId` → workflow never resumes. +- Sending `eventData` that doesn’t match the awaited type. + +--- + +## Realtime Subscription in the Frontend + +Use a single hook to consume all workflow‑related events and manage state. + +```ts +// lib/realtime-client.ts & useWorkflow hooks +useRealtime({ + enabled: !!workflowRunId, + channels: [workflowRunId], + events: [ + "workflow.stepFinish", + "workflow.runFinish", + "workflow.waitingForInput", + "workflow.inputResolved", + ], + onData({ event, data }) { + switch (event) { + case "workflow.stepFinish": + setSteps((prev) => [...prev, data]); + break; + case "workflow.waitingForInput": + setWaitingState(data); + break; + case "workflow.inputResolved": + setWaitingState((prev) => (prev?.eventId === data.eventId ? null : prev)); + break; + case "workflow.runFinish": + setIsRunFinished(true); + } + }, +}); +``` + +Pitfalls: + +- Forgetting to clear waiting state on `inputResolved`. +- Forgetting to reset state on re‑trigger. + +--- + +## Triggering the Workflow + +One simple endpoint triggers both basic and human‑in‑the‑loop flows. + +```ts +// app/api/trigger/route.ts +export async function POST(req: NextRequest) { + const workflowUrl = `${req.nextUrl.origin}/api/workflow`; // or /human-in-loop + const { workflowRunId } = await workflowClient.trigger({ + url: workflowUrl, + body: { userId: "123", action: "process" }, + }); + return NextResponse.json({ workflowRunId }); +} +``` + +--- + +## Recommended Patterns + +- Use one channel per workflow run (`workflowRunId`). +- Always place event emissions inside `context.run()`. +- Use stable `eventId` formats for all interactive steps. +- Subscribe to only the events your UI needs. +- Reset UI state before every new trigger. diff --git a/skills/rest-api.md b/skills/rest-api.md new file mode 100644 index 00000000..6a1d6ef8 --- /dev/null +++ b/skills/rest-api.md @@ -0,0 +1,156 @@ +# Workflow REST API Skill + +This Skill provides TypeScript-focused guidance for interacting with the Upstash QStash Workflow Dead Letter Queue (DLQ) REST API. It includes practical usage examples, consolidated patterns, and key considerations for safely restarting, resuming, listing, deleting, and inspecting failed workflow runs. + +The goal is to help agents construct correct TypeScript API requests, handle optional fields, and avoid common pitfalls when working with workflow recovery operations. + +--- + +## Core Concepts + +• **DLQ Entries** represent failed workflow runs containing metadata such as payload, headers, timestamps, failure callback state, and workflow configuration. +• **Restart** recreates the entire workflow run from the beginning (fresh execution, no preserved state). +• **Resume** creates a new run continuing from the failed step (preserves successful steps). +• **Bulk operations** process up to 50 DLQ entries per request and may return a cursor for pagination. +• **Flow-Control and Retry Overrides** are passed via headers and apply only to the newly created workflow run. + +Be careful: changing workflow code **before** the failed step may cause resume operations to break. + +--- + +# HTTP Endpoints Overview + +Below is a complete summary of the API endpoints supported by this skill. + +• `GET /v2/workflows/dlq` — List DLQ entries with filtering. +• `GET /v2/workflows/dlq/{dlqId}` — Retrieve a single DLQ entry. +• `DELETE /v2/workflows/dlq/{dlqId}` — Delete a DLQ entry. +• `POST /v2/workflows/dlq/restart/{dlqId}` — Restart a workflow from scratch. +• `POST /v2/workflows/dlq/resume/{dlqId}` — Resume a workflow from point of failure. +• `POST /v2/workflows/dlq/restart` — Bulk restart (up to 50). +• `POST /v2/workflows/dlq/resume` — Bulk resume (up to 50). +• `POST /v2/workflows/dlq/callback/{dlqId}` — Rerun a failed failure-callback. +• `GET /v2/flowControl` and `/v2/flowControl/{key}` — Flow-control inspection. + +--- + +# TypeScript Usage Patterns + +Below is a single integrated example illustrating several endpoints, optional query parameters, header overrides, and response field handling. + +```ts +import fetch from "node-fetch"; + +const token = process.env.QSTASH_TOKEN; +const base = "https://qstash.upstash.io/v2"; + +async function example() { + // 1. List DLQ items with filters + const list = await fetch( + `${base}/workflows/dlq?workflowUrl=https://my.app/workflow&fromDate=1700000000000`, + { + headers: { Authorization: `Bearer ${token}` }, + } + ).then((r) => r.json()); + + const firstDlqId = list.messages?.[0]?.dlqId; + + // 2. Restart or Resume a specific run (headers may override flow control or retries) + const restart = await fetch(`${base}/workflows/dlq/restart/${firstDlqId}`, { + method: "POST", + headers: { + Authorization: `Bearer ${token}`, + "Upstash-Flow-Control-Key": "my-key", // optional + "Upstash-Flow-Control-Value": "parallelism=1", // optional + "Upstash-Retries": "3", // optional + }, + }).then((r) => r.json()); + + // 3. Bulk resume using filters and cursor handling + const bulkResume = await fetch(`${base}/workflows/dlq/resume`, { + method: "POST", + headers: { Authorization: `Bearer ${token}` }, + body: JSON.stringify({ + fromDate: 1700000000000, + workflowUrl: "https://my.app/workflow", + dlqIds: [firstDlqId], // optional; filters also supported + }), + }).then((r) => r.json()); + + // 4. Rerun failure callback + const callback = await fetch(`${base}/workflows/dlq/callback/${firstDlqId}`, { + method: "POST", + headers: { Authorization: `Bearer ${token}` }, + }).then((r) => r.json()); + + // 5. Delete a DLQ entry + await fetch(`${base}/workflows/dlq/${firstDlqId}`, { + method: "DELETE", + headers: { Authorization: `Bearer ${token}` }, + }); + + // 6. Inspect Flow-Control + const fc = await fetch(`${base}/flowControl/my-key`, { + headers: { Authorization: `Bearer ${token}` }, + }).then((r) => r.json()); + + console.log({ list, restart, bulkResume, callback, fc }); +} +``` + +--- + +# Key Fields and Behaviors + +## Request Query Fields + +• **dlqIds: string[]** — Exact DLQ IDs for targeting specific failed runs (bulk restart/resume). +• **fromDate / toDate: number** — Millisecond timestamps; inclusive range filters. +• **workflowUrl: string** — Filters by workflow endpoint URL. +• **workflowRunId: string** — Can match full run ID or prefix. +• **workflowCreatedAt: number** — Timestamp filter for creation time. +• **responseStatus: number** — Filter by HTTP status of failed run. +• **callerIP: string** — Filter by origin IP. +• **failureCallbackState: string** — Filter runs whose failure callback succeeded or failed. +• **cursor: string** — Pagination cursor; returned by list & bulk ops. +• **count: number** — Limit for listing DLQ entries (default/max 100). + +## Header Overrides + +These apply only when creating a **new workflow run** (restart or resume). + +• **Upstash-Flow-Control-Key** — Override flow-control key. +• **Upstash-Flow-Control-Value** — Override flow-control config (e.g., "parallelism=1"). +• **Upstash-Retries** — Override step retry configuration. + +All are optional; original values are reused if omitted. + +## Response Fields + +• **cursor?: string** — If returned, additional entries exist. +• **workflowRuns: { workflowRunId: string; workflowCreatedAt: number }[]** — Returned by bulk restart/resume. +• **message objects** (DLQ list/get) include: + +- **workflowRunId, workflowCreatedAt, workflowUrl** +- **responseStatus, responseHeader, responseBody** +- **failureCallbackInfo.state** — Identify failed callbacks. +- **dlqId** — Unique DLQ entry identifier. + +--- + +# Pitfalls & Best Practices + +• **Resume only works if workflow code before the failed step stays unchanged**. Any change may cause resume to fail. +• **Bulk operations process max 50 items**. Always check `cursor` to continue processing. +• **Deleting a DLQ entry is permanent**; once removed, it cannot be resumed or restarted. +• **Failure callback reruns are independent** and do not affect workflow execution. They only retry the failure-notification step. +• **Prefix matching on workflowRunId** can unintentionally match more runs than expected—use full ID for precision. + +--- + +# Recommended Patterns + +• Fetch → Check cursor → Continue looping for bulk operations. +• Always log `failureCallbackInfo.state` to determine whether a callback rerun is required. +• Apply header overrides sparingly; flow-control misconfiguration may stall large batches. +• Prefer filtering in bulk operations instead of manually passing large dlqIds arrays. diff --git a/skills/troubleshooting.md b/skills/troubleshooting.md new file mode 100644 index 00000000..1150741a --- /dev/null +++ b/skills/troubleshooting.md @@ -0,0 +1,201 @@ +# Workflow Troubleshooting (TypeScript) + +This guide provides clear, TypeScript‑focused patterns for diagnosing and fixing common issues in Upstash Workflow. It highlights correct usage of core workflow features and shows how to avoid pitfalls that commonly occur in real applications. + +--- + +## Steps Inside try/catch + +Workflow steps (`context.run`, `context.sleep`, `context.sleepUntil`, `context.call`) intentionally throw `WorkflowAbort` after completing. Catching this error prevents the workflow engine from progressing. + +**Correct pattern:** rethrow `WorkflowAbort` or avoid `try/catch` around steps. + +```ts +import { WorkflowAbort } from "@upstash/workflow"; + +try { + // Any step will throw WorkflowAbort + const result = await context.run("step", () => "ok"); +} catch (err) { + if (err instanceof WorkflowAbort) throw err; // required + // handle real errors +} +``` + +If you _must_ handle errors, move the `try/catch` into the function passed to `context.run`. + +--- + +## requestPayload Becoming Undefined + +During `context.call`, the workflow endpoint is invoked multiple times internally. Before the first step executes, `context.requestPayload` may appear `undefined`. + +**Fix:** capture the payload via a step: + +```ts +export const { POST } = serve(async (context) => { + const payload = await context.run("get-payload", () => context.requestPayload); + + // payload remains stable even during context.call + await context.call("my-call", { + url: "https://example.com", + body: payload, + }); +}); +``` + +**Header considerations:** When triggering workflows, ensure payload‑friendly headers like: + +- `Content-Type: text/plain` +- `Content-Type: application/json` + +--- + +## Signature Verification Failures + +If QStash request verification fails: + +- Ensure the workflow is triggered through QStash (`client.trigger` or publish) +- Verify `QSTASH_CURRENT_SIGNING_KEY` and `QSTASH_NEXT_SIGNING_KEY` +- Pass appropriate `Content-Type` headers when triggering + +Error looks like: + +``` +Failed to verify that the Workflow request comes from QStash: ... +``` + +--- + +## Authorization Errors from Early Returns + +If your workflow returns **before running any step**, the SDK interprets this as failed authorization: + +``` +HTTP 400 – Failed to authenticate Workflow request. +``` + +To safely perform non‑deterministic checks, wrap them in a step: + +```ts +export const { POST } = serve(async (context) => { + const shouldExit = await context.run("check-condition", () => Math.random() > 0.5); + if (shouldExit) return; + + // remaining workflow +}); +``` + +--- + +## Retry Configuration + +Retries come from two locations: + +- Workflow start (`client.trigger`): default **3**, applies to workflow steps +- `context.call`: default **0** +- `context.invoke`: default **0** + +Use these in combination to tune behavior correctly. + +--- + +## Verbose Mode Diagnostics + +Enable verbose logging to debug rare edge cases: + +```ts +serve(handler, { verbose: true }); +``` + +Useful warnings include: + +- Localhost in workflow URL +- Duplicate step execution +- Network response anomalies +- Parallel `context.call` race warnings + +Each of these indicates environmental or routing issues rather than workflow logic problems. + +--- + +## HTTPS Protocol Issues (Proxy Environments) + +If deployed behind a proxy that downgrades HTTPS → HTTP (e.g., Railway), the SDK may infer the wrong protocol. + +**Fix:** explicitly set: + +``` +UPSTASH_WORKFLOW_URL=https://your-deployment-url +``` + +--- + +## Workflow Stuck on First Step + +If the first step appears stuck: + +- Check step logs in the dashboard +- Ensure the endpoint URL is correct and reachable +- Verify SDK versions +- If inference is failing, explicitly set `baseUrl`: + +```ts +serve(handler, { baseUrl: "https://your-url" }); +``` + +--- + +## Non‑workflow Destination Error + +Occurs when triggering a non‑workflow endpoint or using mismatched SDK versions. + +``` +detected a non-workflow destination for trigger/invoke +``` + +**Ensure:** + +- The URL points to a `serve()` workflow endpoint +- Both caller and workflow endpoint use the latest SDK + +--- + +## Vercel Preview Deployment Protection + +Preview deployments block external requests unless bypassed. + +**Steps:** + +1. Create a bypass secret in Vercel → Deployment Protection +2. Add it to the QStash client + pass it when triggering + +```ts +import { Client as QStash } from "@upstash/qstash"; +import { serve } from "@upstash/workflow/nextjs"; + +export const { POST } = serve( + async (context) => { + // workflow logic + }, + { + qstashClient: new QStash({ + token: process.env.QSTASH_TOKEN!, + headers: { + "x-vercel-protection-bypass": process.env.VERCEL_AUTOMATION_BYPASS_SECRET!, + }, + }), + } +); + +// Triggering +await client.trigger({ + url: "https://preview.vercel.app/workflow", + headers: { + "x-vercel-protection-bypass": process.env.VERCEL_AUTOMATION_BYPASS_SECRET!, + }, + body: "hello", +}); +``` + +Both trigger header and client header are required.