WORKFLOW LIVE

Real-time workflow monitoring on the edge

MONITORING PATTERN 2025.01.15

THE WORKFLOW VISIBILITY PROBLEM

Problem: Workflows run in the background with no visibility. You only know they failed when users complain.

Solution: Workflow Live provides real-time monitoring with WebSocket updates, Durable Object rooms per workflow, and Analytics Engine for historical data.

Workflow → DO Room → WebSocket → Live UI
                ↓              ↓              ↓
            Event Model    Real-time Updates  Analytics Engine

PATTERN OVERVIEW

TRADITIONAL APPROACH

  • • Polling for workflow status
  • • No real-time updates
  • • Limited error visibility
  • • Expensive status checks
  • • No historical analytics

WORKFLOW LIVE APPROACH

  • • WebSocket-based live updates
  • • Durable Object rooms per workflow
  • • Real-time error streaming
  • • Cost-effective monitoring
  • • Analytics Engine integration

KEY FEATURES

  • Real-time updates - WebSocket streaming of workflow progress
  • Per-workflow rooms - Durable Object isolation for each workflow
  • Event-driven architecture - Clean separation of concerns
  • Error handling - Automatic retries and failure notifications
  • Analytics integration - Historical data with Analytics Engine
  • Minimal latency - Sub-second update propagation
  • Scalable design - Handles thousands of concurrent workflows

DEMO

Try it live at: workflow-live.coey.dev

Real-time workflow monitoring with live stages, errors, and durations under 1 second.

WORKFLOW MONITOR

Core Workflow Monitoring Logic

// src/lib/workflow-monitor.ts
export interface WorkflowStage {
  id: string;
  name: string;
  status: 'pending' | 'running' | 'completed' | 'failed';
  startTime?: number;
  endTime?: number;
  duration?: number;
  error?: string;
  data?: any;
}

export interface Workflow {
  id: string;
  name: string;
  status: 'pending' | 'running' | 'completed' | 'failed';
  stages: WorkflowStage[];
  startTime: number;
  endTime?: number;
  duration?: number;
  error?: string;
  metadata?: any;
}

export class WorkflowMonitor {
  private workflows = new Map<string, Workflow>();
  private eventHandlers = new Map<string, Set<(workflow: Workflow) => void>>();

  constructor(private env: any) {}

  async createWorkflow(name: string, stages: Omit<WorkflowStage, 'id' | 'status'>[]): Promise<Workflow> {
    const id = crypto.randomUUID();
    const workflow: Workflow = {
      id,
      name,
      status: 'pending',
      stages: stages.map(stage => ({
        ...stage,
        id: crypto.randomUUID(),
        status: 'pending'
      })),
      startTime: Date.now()
    };

    this.workflows.set(id, workflow);
    await this.broadcastUpdate(workflow);
    
    return workflow;
  }

  async startWorkflow(id: string): Promise<Workflow> {
    const workflow = this.workflows.get(id);
    if (!workflow) {
      throw new Error(`Workflow ${id} not found`);
    }

    workflow.status = 'running';
    await this.broadcastUpdate(workflow);
    
    return workflow;
  }

  async updateStage(workflowId: string, stageId: string, updates: Partial<WorkflowStage>): Promise<Workflow> {
    const workflow = this.workflows.get(workflowId);
    if (!workflow) {
      throw new Error(`Workflow ${workflowId} not found`);
    }

    const stage = workflow.stages.find(s => s.id === stageId);
    if (!stage) {
      throw new Error(`Stage ${stageId} not found in workflow ${workflowId}`);
    }

    // Update stage
    Object.assign(stage, updates);
    
    if (updates.status === 'running' && !stage.startTime) {
      stage.startTime = Date.now();
    }
    
    if (updates.status === 'completed' || updates.status === 'failed') {
      stage.endTime = Date.now();
      stage.duration = stage.endTime - (stage.startTime || stage.endTime);
    }

    // Update workflow status based on stages
    this.updateWorkflowStatus(workflow);
    
    await this.broadcastUpdate(workflow);
    
    return workflow;
  }

  async completeWorkflow(id: string, error?: string): Promise<Workflow> {
    const workflow = this.workflows.get(id);
    if (!workflow) {
      throw new Error(`Workflow ${id} not found`);
    }

    workflow.status = error ? 'failed' : 'completed';
    workflow.endTime = Date.now();
    workflow.duration = workflow.endTime - workflow.startTime;
    workflow.error = error;

    await this.broadcastUpdate(workflow);
    
    return workflow;
  }

  getWorkflow(id: string): Workflow | undefined {
    return this.workflows.get(id);
  }

  getAllWorkflows(): Workflow[] {
    return Array.from(this.workflows.values());
  }

  subscribe(workflowId: string, handler: (workflow: Workflow) => void): () => void {
    if (!this.eventHandlers.has(workflowId)) {
      this.eventHandlers.set(workflowId, new Set());
    }
    
    this.eventHandlers.get(workflowId)!.add(handler);
    
    // Return unsubscribe function
    return () => {
      const handlers = this.eventHandlers.get(workflowId);
      if (handlers) {
        handlers.delete(handler);
        if (handlers.size === 0) {
          this.eventHandlers.delete(workflowId);
        }
      }
    };
  }

  private updateWorkflowStatus(workflow: Workflow): void {
    const stages = workflow.stages;
    
    if (stages.every(s => s.status === 'completed')) {
      workflow.status = 'completed';
    } else if (stages.some(s => s.status === 'failed')) {
      workflow.status = 'failed';
    } else if (stages.some(s => s.status === 'running')) {
      workflow.status = 'running';
    } else {
      workflow.status = 'pending';
    }
  }

  private async broadcastUpdate(workflow: Workflow): Promise<void> {
    const handlers = this.eventHandlers.get(workflow.id);
    if (handlers) {
      for (const handler of handlers) {
        try {
          handler(workflow);
        } catch (error) {
          console.error('Error in workflow event handler:', error);
        }
      }
    }

    // Send to Analytics Engine for historical data
    await this.sendToAnalytics(workflow);
  }

  private async sendToAnalytics(workflow: Workflow): Promise<void> {
    try {
      await this.env.ANALYTICS_ENGINE.writeDataPoint({
        blobs: [workflow.id, workflow.name, workflow.status],
        doubles: [workflow.duration || 0, workflow.stages.length],
        indexes: [workflow.name, workflow.status]
      });
    } catch (error) {
      console.error('Failed to send workflow data to Analytics Engine:', error);
    }
  }
}

DURABLE OBJECT ROOM

WebSocket Connection Management

// src/durable-objects/WorkflowRoom.ts
import { DurableObject } from "cloudflare:workers";

export class WorkflowRoom extends DurableObject {
  private connections = new Set<WebSocket>();
  private workflowId: string | null = null;

  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);
    
    if (request.headers.get("Upgrade") === "websocket") {
      return this.handleWebSocket(request);
    }
    
    if (url.pathname === "/join") {
      return this.handleJoin(request);
    }
    
    if (url.pathname === "/broadcast") {
      return this.handleBroadcast(request);
    }

    return new Response("Not found", { status: 404 });
  }

  private async handleWebSocket(request: Request): Promise<Response> {
    const webSocketPair = new WebSocketPair();
    const [client, server] = Object.values(webSocketPair);

    server.accept();
    this.connections.add(server);

    // Handle WebSocket events
    server.addEventListener("message", (event) => {
      try {
        const data = JSON.parse(event.data as string);
        this.handleWebSocketMessage(server, data);
      } catch (error) {
        console.error("Error parsing WebSocket message:", error);
        server.send(JSON.stringify({ error: "Invalid message format" }));
      }
    });

    server.addEventListener("close", () => {
      this.connections.delete(server);
    });

    server.addEventListener("error", (error) => {
      console.error("WebSocket error:", error);
      this.connections.delete(server);
    });

    // Send welcome message
    server.send(JSON.stringify({
      type: "connected",
      workflowId: this.workflowId,
      timestamp: Date.now()
    }));

    return new Response(null, {
      status: 101,
      webSocket: client
    });
  }

  private async handleJoin(request: Request): Promise<Response> {
    const { workflowId } = await request.json();
    
    if (!workflowId) {
      return new Response("Missing workflowId", { status: 400 });
    }

    this.workflowId = workflowId;
    
    // Broadcast join event to all connections
    this.broadcast({
      type: "workflow_joined",
      workflowId,
      timestamp: Date.now()
    });

    return new Response(JSON.stringify({ success: true }), {
      headers: { "Content-Type": "application/json" }
    });
  }

  private async handleBroadcast(request: Request): Promise<Response> {
    const data = await request.json();
    
    this.broadcast({
      type: "workflow_update",
      data,
      timestamp: Date.now()
    });

    return new Response(JSON.stringify({ success: true }), {
      headers: { "Content-Type": "application/json" }
    });
  }

  private handleWebSocketMessage(server: WebSocket, data: any): void {
    switch (data.type) {
      case "ping":
        server.send(JSON.stringify({ type: "pong", timestamp: Date.now() }));
        break;
      case "subscribe":
        // Handle subscription to specific workflow events
        break;
      default:
        server.send(JSON.stringify({ error: "Unknown message type" }));
    }
  }

  private broadcast(message: any): void {
    const messageStr = JSON.stringify(message);
    
    for (const connection of this.connections) {
      try {
        connection.send(messageStr);
      } catch (error) {
        console.error("Error broadcasting message:", error);
        this.connections.delete(connection);
      }
    }
  }

  // Called by external systems to broadcast workflow updates
  async broadcastWorkflowUpdate(workflow: any): Promise<void> {
    this.broadcast({
      type: "workflow_update",
      workflow,
      timestamp: Date.now()
    });
  }

  // Called by external systems to broadcast stage updates
  async broadcastStageUpdate(workflowId: string, stage: any): Promise<void> {
    this.broadcast({
      type: "stage_update",
      workflowId,
      stage,
      timestamp: Date.now()
    });
  }

  // Called by external systems to broadcast errors
  async broadcastError(workflowId: string, error: any): Promise<void> {
    this.broadcast({
      type: "workflow_error",
      workflowId,
      error,
      timestamp: Date.now()
    });
  }
}

WORKER IMPLEMENTATION

HTTP/WebSocket Routing and Workflow Execution

// src/worker.ts
import { WorkflowRoom } from "./durable-objects/WorkflowRoom";
import { WorkflowMonitor } from "./lib/workflow-monitor";

export default {
  async fetch(request: Request, env: any): Promise<Response> {
    const url = new URL(request.url);
    
    // CORS headers
    const corsHeaders = {
      "Access-Control-Allow-Origin": "*",
      "Access-Control-Allow-Methods": "GET, POST, PUT, DELETE, OPTIONS",
      "Access-Control-Allow-Headers": "Content-Type, Authorization"
    };

    if (request.method === "OPTIONS") {
      return new Response(null, { headers: corsHeaders });
    }

    try {
      // Route to appropriate handler
      if (url.pathname.startsWith("/workflow/")) {
        return handleWorkflowRequest(request, env, corsHeaders);
      } else if (url.pathname.startsWith("/room/")) {
        return handleRoomRequest(request, env, corsHeaders);
      } else if (url.pathname === "/") {
        return new Response("Workflow Live API", { headers: corsHeaders });
      }

      return new Response("Not found", { status: 404, headers: corsHeaders });
    } catch (error) {
      console.error("Worker error:", error);
      return new Response("Internal server error", { 
        status: 500, 
        headers: corsHeaders 
      });
    }
  }
};

async function handleWorkflowRequest(request: Request, env: any, corsHeaders: any): Promise<Response> {
  const url = new URL(request.url);
  const pathParts = url.pathname.split("/");
  const workflowId = pathParts[2];
  const action = pathParts[3];

  const monitor = new WorkflowMonitor(env);

  switch (request.method) {
    case "POST":
      if (action === "create") {
        const { name, stages } = await request.json();
        const workflow = await monitor.createWorkflow(name, stages);
        return new Response(JSON.stringify(workflow), {
          headers: { ...corsHeaders, "Content-Type": "application/json" }
        });
      } else if (action === "start") {
        const workflow = await monitor.startWorkflow(workflowId);
        return new Response(JSON.stringify(workflow), {
          headers: { ...corsHeaders, "Content-Type": "application/json" }
        });
      }
      break;

    case "PUT":
      if (action === "stage") {
        const { stageId, updates } = await request.json();
        const workflow = await monitor.updateStage(workflowId, stageId, updates);
        
        // Broadcast to room
        const roomId = env.WORKFLOW_ROOM.idFromName(workflowId);
        const room = env.WORKFLOW_ROOM.get(roomId);
        await room.broadcastStageUpdate(workflowId, updates);
        
        return new Response(JSON.stringify(workflow), {
          headers: { ...corsHeaders, "Content-Type": "application/json" }
        });
      } else if (action === "complete") {
        const { error } = await request.json();
        const workflow = await monitor.completeWorkflow(workflowId, error);
        
        // Broadcast to room
        const roomId = env.WORKFLOW_ROOM.idFromName(workflowId);
        const room = env.WORKFLOW_ROOM.get(roomId);
        await room.broadcastWorkflowUpdate(workflow);
        
        return new Response(JSON.stringify(workflow), {
          headers: { ...corsHeaders, "Content-Type": "application/json" }
        });
      }
      break;

    case "GET":
      if (action === "status") {
        const workflow = monitor.getWorkflow(workflowId);
        if (!workflow) {
          return new Response("Workflow not found", { 
            status: 404, 
            headers: corsHeaders 
          });
        }
        return new Response(JSON.stringify(workflow), {
          headers: { ...corsHeaders, "Content-Type": "application/json" }
        });
      } else if (!action) {
        // Get all workflows
        const workflows = monitor.getAllWorkflows();
        return new Response(JSON.stringify(workflows), {
          headers: { ...corsHeaders, "Content-Type": "application/json" }
        });
      }
      break;
  }

  return new Response("Method not allowed", { status: 405, headers: corsHeaders });
}

async function handleRoomRequest(request: Request, env: any, corsHeaders: any): Promise<Response> {
  const url = new URL(request.url);
  const pathParts = url.pathname.split("/");
  const workflowId = pathParts[2];
  const action = pathParts[3];

  if (!workflowId) {
    return new Response("Missing workflow ID", { status: 400, headers: corsHeaders });
  }

  const roomId = env.WORKFLOW_ROOM.idFromName(workflowId);
  const room = env.WORKFLOW_ROOM.get(roomId);

  if (action === "join") {
    return room.fetch(new Request("/join", {
      method: "POST",
      body: JSON.stringify({ workflowId })
    }));
  } else if (action === "websocket") {
    return room.fetch(request);
  }

  return new Response("Not found", { status: 404, headers: corsHeaders });
}

FRONTEND COMPONENT

Real-time UI with WebSocket Updates

<!-- src/static/index.html -->
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Workflow Live Monitor</title>
    <style>
        body {
            font-family: 'Courier New', monospace;
            background: #000;
            color: #0f0;
            margin: 0;
            padding: 20px;
            line-height: 1.6;
        }
        
        .container {
            max-width: 1200px;
            margin: 0 auto;
        }
        
        .header {
            text-align: center;
            margin-bottom: 30px;
            border-bottom: 2px solid #0f0;
            padding-bottom: 20px;
        }
        
        .workflow-list {
            display: grid;
            gap: 20px;
            margin-bottom: 30px;
        }
        
        .workflow-card {
            border: 1px solid #0f0;
            padding: 20px;
            background: #001100;
        }
        
        .workflow-header {
            display: flex;
            justify-content: space-between;
            align-items: center;
            margin-bottom: 15px;
        }
        
        .workflow-name {
            font-size: 1.2em;
            font-weight: bold;
        }
        
        .workflow-status {
            padding: 5px 10px;
            border: 1px solid #0f0;
            background: #002200;
        }
        
        .workflow-status.running {
            background: #004400;
            animation: pulse 2s infinite;
        }
        
        .workflow-status.completed {
            background: #006600;
        }
        
        .workflow-status.failed {
            background: #660000;
            color: #ff0000;
        }
        
        @keyframes pulse {
            0%, 100% { opacity: 1; }
            50% { opacity: 0.5; }
        }
        
        .stages {
            margin-top: 15px;
        }
        
        .stage {
            display: flex;
            align-items: center;
            margin: 8px 0;
            padding: 8px;
            border-left: 3px solid #333;
        }
        
        .stage.running {
            border-left-color: #0f0;
            background: #001100;
        }
        
        .stage.completed {
            border-left-color: #0f0;
            background: #002200;
        }
        
        .stage.failed {
            border-left-color: #f00;
            background: #220000;
        }
        
        .stage-name {
            flex: 1;
        }
        
        .stage-duration {
            color: #666;
            font-size: 0.9em;
        }
        
        .controls {
            text-align: center;
            margin-bottom: 30px;
        }
        
        button {
            background: #000;
            color: #0f0;
            border: 1px solid #0f0;
            padding: 10px 20px;
            margin: 0 10px;
            cursor: pointer;
            font-family: inherit;
        }
        
        button:hover {
            background: #001100;
        }
        
        button:disabled {
            opacity: 0.5;
            cursor: not-allowed;
        }
        
        .error {
            color: #f00;
            background: #220000;
            padding: 10px;
            border: 1px solid #f00;
            margin: 10px 0;
        }
        
        .connection-status {
            position: fixed;
            top: 10px;
            right: 10px;
            padding: 5px 10px;
            border: 1px solid #0f0;
            background: #001100;
        }
        
        .connection-status.disconnected {
            border-color: #f00;
            background: #220000;
            color: #f00;
        }
    </style>
</head>
<body>
    <div class="container">
        <div class="header">
            <h1>🚀 WORKFLOW LIVE MONITOR</h1>
            <p>Real-time workflow monitoring powered by Cloudflare Workers & Durable Objects</p>
        </div>
        
        <div class="connection-status" id="connectionStatus">
            Connecting...
        </div>
        
        <div class="controls">
            <button onclick="createSampleWorkflow()">Create Sample Workflow</button>
            <button onclick="startAllWorkflows()">Start All Workflows</button>
            <button onclick="clearWorkflows()">Clear All</button>
        </div>
        
        <div id="errorContainer"></div>
        
        <div class="workflow-list" id="workflowList">
            <!-- Workflows will be populated here -->
        </div>
    </div>

    <script>
        class WorkflowMonitor {
            constructor() {
                this.workflows = new Map();
                this.websocket = null;
                this.connectionStatus = document.getElementById('connectionStatus');
                this.workflowList = document.getElementById('workflowList');
                this.errorContainer = document.getElementById('errorContainer');
                
                this.connect();
            }
            
            connect() {
                try {
                    const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
                    const wsUrl = `${protocol}//${window.location.host}/room/${crypto.randomUUID()}/websocket`;
                    
                    this.websocket = new WebSocket(wsUrl);
                    
                    this.websocket.onopen = () => {
                        this.connectionStatus.textContent = 'Connected';
                        this.connectionStatus.className = 'connection-status';
                    };
                    
                    this.websocket.onmessage = (event) => {
                        const data = JSON.parse(event.data);
                        this.handleMessage(data);
                    };
                    
                    this.websocket.onclose = () => {
                        this.connectionStatus.textContent = 'Disconnected';
                        this.connectionStatus.className = 'connection-status disconnected';
                        
                        // Reconnect after 3 seconds
                        setTimeout(() => this.connect(), 3000);
                    };
                    
                    this.websocket.onerror = (error) => {
                        console.error('WebSocket error:', error);
                        this.showError('WebSocket connection error');
                    };
                    
                } catch (error) {
                    console.error('Failed to connect:', error);
                    this.showError('Failed to connect to workflow monitor');
                }
            }
            
            handleMessage(data) {
                switch (data.type) {
                    case 'connected':
                        console.log('Connected to workflow monitor');
                        break;
                    case 'workflow_update':
                        this.updateWorkflow(data.workflow);
                        break;
                    case 'stage_update':
                        this.updateStage(data.workflowId, data.stage);
                        break;
                    case 'workflow_error':
                        this.showError(`Workflow ${data.workflowId} error: ${data.error.message}`);
                        break;
                }
            }
            
            async createSampleWorkflow() {
                try {
                    const response = await fetch('/workflow/create', {
                        method: 'POST',
                        headers: { 'Content-Type': 'application/json' },
                        body: JSON.stringify({
                            name: `Sample Workflow ${Date.now()}`,
                            stages: [
                                { name: 'Initialize' },
                                { name: 'Process Data' },
                                { name: 'Validate Results' },
                                { name: 'Send Notifications' },
                                { name: 'Cleanup' }
                            ]
                        })
                    });
                    
                    if (!response.ok) {
                        throw new Error(`HTTP ${response.status}: ${response.statusText}`);
                    }
                    
                    const workflow = await response.json();
                    this.workflows.set(workflow.id, workflow);
                    this.renderWorkflows();
                    
                    // Join the workflow room
                    await this.joinWorkflowRoom(workflow.id);
                    
                } catch (error) {
                    this.showError(`Failed to create workflow: ${error.message}`);
                }
            }
            
            async joinWorkflowRoom(workflowId) {
                try {
                    await fetch(`/room/${workflowId}/join`, {
                        method: 'POST',
                        headers: { 'Content-Type': 'application/json' },
                        body: JSON.stringify({ workflowId })
                    });
                } catch (error) {
                    console.error('Failed to join workflow room:', error);
                }
            }
            
            async startAllWorkflows() {
                for (const [id, workflow] of this.workflows) {
                    if (workflow.status === 'pending') {
                        await this.startWorkflow(id);
                    }
                }
            }
            
            async startWorkflow(workflowId) {
                try {
                    const response = await fetch(`/workflow/${workflowId}/start`, {
                        method: 'POST'
                    });
                    
                    if (!response.ok) {
                        throw new Error(`HTTP ${response.status}: ${response.statusText}`);
                    }
                    
                    const workflow = await response.json();
                    this.workflows.set(workflowId, workflow);
                    this.renderWorkflows();
                    
                    // Simulate workflow execution
                    this.simulateWorkflowExecution(workflowId);
                    
                } catch (error) {
                    this.showError(`Failed to start workflow: ${error.message}`);
                }
            }
            
            simulateWorkflowExecution(workflowId) {
                const workflow = this.workflows.get(workflowId);
                if (!workflow) return;
                
                let currentStageIndex = 0;
                
                const executeNextStage = async () => {
                    if (currentStageIndex >= workflow.stages.length) {
                        // Complete workflow
                        await this.completeWorkflow(workflowId);
                        return;
                    }
                    
                    const stage = workflow.stages[currentStageIndex];
                    
                    // Start stage
                    await this.updateStage(workflowId, stage.id, { status: 'running' });
                    
                    // Simulate work (random duration between 1-3 seconds)
                    const duration = Math.random() * 2000 + 1000;
                    
                    setTimeout(async () => {
                        // Complete stage (with occasional failure)
                        const shouldFail = Math.random() < 0.1; // 10% failure rate
                        
                        if (shouldFail) {
                            await this.updateStage(workflowId, stage.id, { 
                                status: 'failed', 
                                error: 'Simulated failure' 
                            });
                            await this.completeWorkflow(workflowId, 'Stage failed');
                        } else {
                            await this.updateStage(workflowId, stage.id, { status: 'completed' });
                            currentStageIndex++;
                            executeNextStage();
                        }
                    }, duration);
                };
                
                executeNextStage();
            }
            
            async updateStage(workflowId, stageId, updates) {
                try {
                    const response = await fetch(`/workflow/${workflowId}/stage`, {
                        method: 'PUT',
                        headers: { 'Content-Type': 'application/json' },
                        body: JSON.stringify({ stageId, updates })
                    });
                    
                    if (!response.ok) {
                        throw new Error(`HTTP ${response.status}: ${response.statusText}`);
                    }
                    
                    const workflow = await response.json();
                    this.workflows.set(workflowId, workflow);
                    this.renderWorkflows();
                    
                } catch (error) {
                    this.showError(`Failed to update stage: ${error.message}`);
                }
            }
            
            async completeWorkflow(workflowId, error = null) {
                try {
                    const response = await fetch(`/workflow/${workflowId}/complete`, {
                        method: 'PUT',
                        headers: { 'Content-Type': 'application/json' },
                        body: JSON.stringify({ error })
                    });
                    
                    if (!response.ok) {
                        throw new Error(`HTTP ${response.status}: ${response.statusText}`);
                    }
                    
                    const workflow = await response.json();
                    this.workflows.set(workflowId, workflow);
                    this.renderWorkflows();
                    
                } catch (error) {
                    this.showError(`Failed to complete workflow: ${error.message}`);
                }
            }
            
            updateWorkflow(workflow) {
                this.workflows.set(workflow.id, workflow);
                this.renderWorkflows();
            }
            
            renderWorkflows() {
                this.workflowList.innerHTML = '';
                
                for (const [id, workflow] of this.workflows) {
                    const workflowCard = document.createElement('div');
                    workflowCard.className = 'workflow-card';
                    
                    const stagesHtml = workflow.stages.map(stage => `
                        <div class="stage ${stage.status}">
                            <div class="stage-name">${stage.name}</div>
                            <div class="stage-duration">
                                ${stage.duration ? `${stage.duration}ms` : ''}
                                ${stage.error ? ` - ${stage.error}` : ''}
                            </div>
                        </div>
                    `).join('');
                    
                    workflowCard.innerHTML = `
                        <div class="workflow-header">
                            <div class="workflow-name">${workflow.name}</div>
                            <div class="workflow-status ${workflow.status}">${workflow.status.toUpperCase()}</div>
                        </div>
                        <div class="stages">
                            ${stagesHtml}
                        </div>
                    `;
                    
                    this.workflowList.appendChild(workflowCard);
                }
            }
            
            showError(message) {
                const errorDiv = document.createElement('div');
                errorDiv.className = 'error';
                errorDiv.textContent = message;
                this.errorContainer.appendChild(errorDiv);
                
                // Remove error after 5 seconds
                setTimeout(() => {
                    if (errorDiv.parentNode) {
                        errorDiv.parentNode.removeChild(errorDiv);
                    }
                }, 5000);
            }
            
            clearWorkflows() {
                this.workflows.clear();
                this.renderWorkflows();
            }
        }
        
        // Initialize the monitor
        const monitor = new WorkflowMonitor();
        
        // Global functions for buttons
        window.createSampleWorkflow = () => monitor.createSampleWorkflow();
        window.startAllWorkflows = () => monitor.startAllWorkflows();
        window.clearWorkflows = () => monitor.clearWorkflows();
    </script>
</body>
</html>

WRANGLER CONFIG

Cloudflare Workers Configuration

# wrangler.toml
name = "workflow-live"
main = "src/worker.ts"
compatibility_date = "2024-01-15"
compatibility_flags = ["nodejs_compat"]

[durable_objects]
bindings = [
  { name = "WORKFLOW_ROOM", class_name = "WorkflowRoom" }
]

[[durable_objects.migrations]]
tag = "v1"
new_classes = ["WorkflowRoom"]

[analytics_engine_datasets]
bindings = [
  { name = "ANALYTICS_ENGINE", dataset = "workflow_analytics" }
]

[vars]
ENVIRONMENT = "production"

# Development
[env.development]
vars = { ENVIRONMENT = "development" }

# Production
[env.production]
vars = { ENVIRONMENT = "production" }

ARCHITECTURE PATTERNS

WORKFLOW EXECUTION

  • • Worker handles HTTP/WebSocket routing
  • • Manages workflow execution
  • • Broadcasts updates via Durable Object
  • • Handles failures gracefully

CONNECTION MANAGEMENT

  • • Durable Object manages WebSocket connections
  • • Handles broadcast messaging
  • • Maintains connection state
  • • Per-workflow room isolation

REAL-TIME UPDATES

  • • Simple, accessible UI
  • • Real-time status updates
  • • Error handling and retries
  • • Visual progress feedback

ANALYTICS INTEGRATION

  • • Analytics Engine for history
  • • Event model for data collection
  • • Historical workflow analysis
  • • Performance metrics tracking

WORKFLOW TYPES

  • AI Agent workflows - Monitor agent execution and decision making
  • Data processing pipelines - Track ETL jobs and transformations
  • User onboarding flows - Monitor multi-step user journeys
  • Background jobs - Track long-running tasks and cleanup
  • Integration workflows - Monitor API calls and data sync
  • Deployment pipelines - Track build and deployment stages
  • Business processes - Monitor approval workflows and notifications

PERFORMANCE CHARACTERISTICS

  • Update latency: <100ms for WebSocket events
  • Workflow execution: Sub-second stage transitions
  • Connection handling: 1000+ concurrent WebSocket connections
  • Error detection: Real-time failure notifications
  • Analytics ingestion: High-throughput event streaming
  • Global distribution: 200+ edge locations

USE CASES

  • AI Agent monitoring and control with real-time decision tracking
  • Edge computing mini-apps with live status updates
  • Real-time workflow orchestration for complex business processes
  • Live system monitoring with instant failure alerts
  • Interactive edge applications with user feedback
  • Data pipeline monitoring with stage-by-stage visibility
  • Multi-step form processing with progress tracking

QUICK START

# Clone the repo
git clone https://github.com/acoyfellow/workflow-live
cd workflow-live

# Install dependencies 
bun install

# Configure Cloudflare
wrangler login

# Deploy
bun run deploy

Creates a complete workflow monitoring system with real-time updates and analytics.

DEVELOPMENT WORKFLOW

  1. Local Development: bun run dev starts local Workers and WebSocket server
  2. Workflow Definition: Define workflow stages and error handling
  3. Testing: bun run test runs unit and integration tests
  4. Deployment: bun run deploy pushes to Cloudflare
  5. Monitoring: Real-time metrics in Cloudflare dashboard

Related patterns