Wait For Event
Notify
You can notify all the workflow runs waitingi for a specific event ID. There are two ways to send a notify request.
Notify within Workflow
Notifies other workflows waiting for a specific event from within a workflow.
TypeScript
import { serve } from "@upstash/workflow/nextjs";export const { POST } = serve<string>(async (context) => { const { orderId, processingResult } = context.requestPayload; await context.run("process-order", async () => { // ... }) const { notifyResponse } = await context.notify( "notify-processing-complete", `order-${orderId}`, { orderId, status: "completed", result: processingResult, completedAt: new Date().toISOString() } );});Python
from fastapi import FastAPIfrom upstash_workflow.fastapi import Servefrom upstash_workflow import AsyncWorkflowContextfrom datetime import datetimeapp = FastAPI()serve = Serve(app)@serve.post("/api/order-processor")async def order_processor(context: AsyncWorkflowContext[str]) -> None: order_id = context.request_payload["order_id"] processing_result = context.request_payload["processing_result"] # Process the order async def _process_order(): return await process_order(order_id) result = await context.run("process-order", _process_order) # Notify waiting workflows that processing is complete notify_response = await context.notify( "notify-processing-complete", f"order-{order_id}", { "order_id": order_id, "status": "completed", "result": processing_result, "completed_at": datetime.utcnow().isoformat() } ) # Log notification results async def _log_notification(): print(f"Notified {len(notify_response)} waiting workflows") return notify_response await context.run("log-notification", _log_notification)External Notification
You can also notify workflows from external systems using the Workflow Client:
TypeScript
import { Client } from "@upstash/workflow";const client = new Client({ token: "<WORKFLOW_TOKEN>" });await client.notify({ eventId: "order-completed-123", eventData: { orderId: "123", status: "completed", deliveryTime: "2 days", trackingNumber: "TRK123456" }});Python
from upstash_workflow import Clientclient = Client("<WORKFLOW_TOKEN>")# Notify workflows waiting for a specific eventawait client.notify( event_id="order-completed-123", event_data={ "order_id": "123", "status": "completed", "delivery_time": "2 days", "tracking_number": "TRK123456" })Lookback Functionality
By default, if you call notify before a workflow reaches its waitForEvent step, the notification will be lost (race condition). To prevent this, you can provide a workflowRunId parameter which enables lookback - the notification will be stored and delivered even if sent before the wait step.
This is particularly useful when:
- You trigger a workflow and immediately want to send it an event
- You have concurrent operations where timing is unpredictable
- You want to eliminate race conditions in your event-driven workflows
TypeScript
import { Client } from "@upstash/workflow";const client = new Client({ token: "<WORKFLOW_TOKEN>" });// Trigger a workflow and get its run IDconst { workflowRunId } = await client.trigger({ url: "https://your-app.com/api/process-order", body: { orderId: "123" }});// Immediately notify it with lookback enabled// The notification will be delivered even if the workflow// hasn't reached waitForEvent yetawait client.notify({ eventId: "payment-verified", eventData: { verified: true, amount: 100 }, workflowRunId: workflowRunId, // Enables lookback});Python
from upstash_workflow import Clientclient = Client("<WORKFLOW_TOKEN>")# Trigger a workflow and get its run IDworkflow_run = await client.trigger( url="https://your-app.com/api/process-order", body={"order_id": "123"})# Immediately notify it with lookback enabledawait client.notify( event_id="payment-verified", event_data={"verified": True, "amount": 100}, workflow_run_id=workflow_run.workflow_run_id # Enables lookback)The same also applies to context.notify
TypeScript
import { serve } from "@upstash/workflow/nextjs";export const { POST } = serve<string>(async (context) => { const { orderId, processingResult } = context.requestPayload; await context.run("process-order", async () => { // ... }) const { notifyResponse } = await context.notify( "notify-processing-complete", `order-${orderId}`, { orderId, status: "completed", result: processingResult, completedAt: new Date().toISOString(), workflowRunId: "targetWorkflowRunId" // Enables lookback } );});When using lookback with workflowRunId, the notification is targeted to a specific workflow run rather than all waiters with that event ID.