Upstash Workflow

Upstash Workflow Implementation Guide

Standard patterns for implementing Upstash Workflow + QStash async workflows in the LobeHub codebase.

🎯 The Three Core Patterns

Every workflow in LobeHub combines these three patterns. They exist because the platform constrains you in three ways: rate limits make blind fan-out dangerous, step limits cap a single workflow's size, and idempotency demands that retries don't double-process.

1. 🔍 Dry-Run Mode — get statistics without triggering actual execution 2. 🌟 Fan-Out Pattern — split large batches into smaller chunks for parallel processing 3. 🎯 Single Task Execution — each workflow execution processes exactly ONE item

---

Architecture Overview

All workflows follow the same 3-layer architecture:

Layer 1: Entry Point (process-*)
  ├─ Validates prerequisites
  ├─ Calculates total items to process
  ├─ Filters existing items
  ├─ Supports dry-run mode (statistics only)
  └─ Triggers Layer 2 if work is needed

Layer 2: Pagination (paginate-*)
  ├─ Handles cursor-based pagination
  ├─ Implements fan-out for large batches
  ├─ Recursively processes all pages
  └─ Triggers Layer 3 for each item

Layer 3: Single Task Execution (execute-* / generate-*)
  └─ Performs actual business logic for ONE item

Real examples in this codebase: welcome-placeholder, agent-welcome — see `references/examples.md`.

---

The Three Patterns in 60 Seconds

1. Dry-Run Mode

Short-circuit Layer 1 before any side effects so callers can preview what would happen:

if (dryRun) {
  return {
    ...result,
    dryRun: true,
    message: `[DryRun] Would process ${itemsNeedingProcessing.length} items`,
  };
}

Use case: check how many items will be processed before committing.

2. Fan-Out Pattern

Layer 2 splits oversized batches into chunks and recursively re-triggers itself with each chunk. This avoids hitting workflow step limits when one page contains too many items:

const CHUNK_SIZE = 20;

if (itemIds.length > CHUNK_SIZE) {
  const chunks = chunk(itemIds, CHUNK_SIZE);
  await Promise.all(
    chunks.map((ids, idx) =>
      context.run(`workflow:fanout:${idx + 1}/${chunks.length}`, () =>
        WorkflowClass.triggerPaginateItems({ itemIds: ids }),
      ),
    ),
  );
}

Defaults: PAGE_SIZE = 50 (items per page), CHUNK_SIZE = 20 (items per fan-out chunk).

3. Single Task Execution

Layer 3 always processes exactly one item per invocation. Parallelism comes from Layer 2 fanning out to many Layer 3 invocations, controlled by flowControl:

export const { POST } = serve<ExecutePayload>(
  async (context) => {
    const { itemId } = context.requestPayload ?? {};
    if (!itemId) return { success: false, error: 'Missing itemId' };

    const item = await context.run('workflow:get-item', () => getItem(itemId));
    const result = await context.run('workflow:execute', () => processItem(item));
    await context.run('workflow:save', () => saveResult(itemId, result));

    return { success: true, itemId, result };
  },
  {
    flowControl: { key: 'workflow.execute', parallelism: 10, ratePerSecond: 5 },
  },
);

---

File Structure

src/
├── app/(backend)/api/workflows/
│   └── {workflow-name}/
│       ├── process-{entities}/route.ts      # Layer 1
│       ├── paginate-{entities}/route.ts     # Layer 2
│       └── execute-{entity}/route.ts        # Layer 3
│
└── server/workflows/
    └── {workflowName}/
        └── index.ts                          # Workflow class

---

Where to Go Next

Pick the reference that matches what you're doing:

---

Environment Variables

# Required for all workflows
APP_URL=https://your-app.com # Base URL for workflow endpoints
QSTASH_TOKEN=qstash_xxx      # QStash authentication token

# Optional (for custom QStash URL)
QSTASH_URL=https://custom-qstash.com

---

Checklist for New Workflows

Planning

• [ ] Identify the entity to process (users, agents, items, …)
• [ ] Define the per-item business logic
• [ ] Determine filtering logic (Redis cache, database state, …)

Implementation

• [ ] Define payload types with TypeScript interfaces
• [ ] Create workflow class with static trigger methods
• [ ] Layer 1: entry point with dry-run support
• [ ] Layer 1: filtering logic to avoid duplicate work
• [ ] Layer 2: pagination with fan-out
• [ ] Layer 3: single-task execution (ONE item per run)
• [ ] Configure appropriate flowControl for each layer
• [ ] Consistent logging with workflow prefixes
• [ ] Validate all required payload parameters
• [ ] Unique context.run() step names

Quality & Deployment

• [ ] Return consistent response shapes
• [ ] Configure cloud deployment (`references/cloud.md` if on lobehub-cloud)
• [ ] Write integration tests (dryRun path + full path)
• [ ] Smoke-test with dry-run first
• [ ] Test with a small batch before full rollout

---

Additional Resources

• Upstash Workflow Documentation
• QStash Documentation
• Example Workflows in Codebase/api/workflows/>)
• Workflow Classes

Best Practices & Common Pitfalls

Apply these once your scaffold from implementation.md is in place.

Table of Contents

1. Error Handling 2. Logging 3. Return Values 4. flowControl Configuration 5. context.run() Best Practices 6. Payload Validation 7. Database Connection 8. Testing 9. Common Pitfalls

---

1. Error Handling

export const { POST } = serve<Payload>(
  async (context) => {
    const { itemId } = context.requestPayload ?? {};

    if (!itemId) {
      return { success: false, error: 'Missing itemId in payload' };
    }

    try {
      const result = await context.run('step-name', () => doWork(itemId));
      return { success: true, itemId, result };
    } catch (error) {
      console.error('[workflow:error]', error);
      return {
        success: false,
        error: error instanceof Error ? error.message : 'Unknown error',
      };
    }
  },
  { flowControl: { ... } },
);

2. Logging

Consistent prefixes make debugging much easier across QStash dashboards and grep:

console.log('[{workflow}:{layer}] Starting with payload:', payload);
console.log('[{workflow}:{layer}] Processing items:', { count: items.length });
console.log('[{workflow}:{layer}] Completed:', result);
console.error('[{workflow}:{layer}:error]', error);

3. Return Values

Pick the shape that matches the layer's purpose — entry points return statistics, execution layers return per-item results.

// Success
return { success: true, itemId, result, message: 'Optional success message' };

// Error
return { success: false, error: 'Error description', itemId };

// Statistics (entry point)
return {
  success: true,
  totalEligible: 100,
  toProcess: 80,
  alreadyProcessed: 20,
  dryRun: true, // if applicable
  message: 'Summary message',
};

4. flowControl Configuration

Tune concurrency by layer — entry points are singletons, execution layers fan out.

// Layer 1: Entry — single instance to avoid duplicate processing
flowControl: { key: '{workflow}.process',  parallelism: 1,  ratePerSecond: 1 }

// Layer 2: Pagination — moderate concurrency
flowControl: { key: '{workflow}.paginate', parallelism: 20, ratePerSecond: 5 }

// Layer 3: Execution — higher concurrency for parallel item work
flowControl: { key: '{workflow}.execute',  parallelism: 10, ratePerSecond: 5 }

Why these defaults:

• Layer 1 always uses parallelism: 1 so concurrent triggers don't both start the same batch.
• Layer 2 can fan out widely (10-20) since pagination is cheap.
• Layer 3 caps at 5-10 by default; raise/lower based on external API rate limits.

5. context.run() Best Practices

• Use descriptive step names with prefixes: {workflow}:step-name
• Each step should be idempotent (safe to retry)
• Don't nest context.run() calls — keep them flat
• Use unique step names when processing multiple items:

// ✅ Unique step names
await Promise.all(
  items.map((item) => context.run(`{workflow}:execute:${item.id}`, () => processItem(item))),
);

// ❌ Same step name — Upstash de-dupes by step name and you'll lose data
await Promise.all(items.map((item) => context.run(`{workflow}:execute`, () => processItem(item))));

6. Payload Validation

Validate at the top so failures are explicit, not silent undefined cascades:

export const { POST } = serve<Payload>(
  async (context) => {
    const { itemId, configId } = context.requestPayload ?? {};

    if (!itemId)   return { success: false, error: 'Missing itemId in payload' };
    if (!configId) return { success: false, error: 'Missing configId in payload' };

    // Proceed with work...
  },
  { flowControl: { ... } },
);

7. Database Connection

Get the connection once per workflow — getServerDB() is async, repeating it inside each step adds latency:

export const { POST } = serve<Payload>(
  async (context) => {
    const db = await getServerDB();

    const item = await context.run('get-item', () => itemModel.findById(db, itemId));
    const result = await context.run('save-result', () => resultModel.create(db, result));
  },
  { flowControl: { ... } },
);

8. Testing

Integration tests should exercise both the dry-run statistics path and the full execution path:

describe('WorkflowName', () => {
  it('should process items successfully', async () => {
    const items = await createTestItems();
    await WorkflowClass.triggerProcessItems({ dryRun: false });
    await waitForCompletion();
    const results = await getResults();
    expect(results).toHaveLength(items.length);
  });

  it('should support dryRun mode', async () => {
    const result = await WorkflowClass.triggerProcessItems({ dryRun: true });
    expect(result).toMatchObject({
      success: true,
      dryRun: true,
      totalEligible: expect.any(Number),
      toProcess: expect.any(Number),
    });
  });
});

---

Common Pitfalls

❌ Reusing context.run() step names

// Bad — Upstash dedupes by step name
await Promise.all(items.map((item) => context.run('process', () => process(item))));

// Good
await Promise.all(items.map((item) => context.run(`process:${item.id}`, () => process(item))));

❌ Skipping payload validation

// Bad — undefined cascades into a confusing failure later
const { itemId } = context.requestPayload ?? {};
const result = await process(itemId);

// Good — fail fast with a clear error
if (!itemId) return { success: false, error: 'Missing itemId' };

❌ Skipping the filter step

// Bad — duplicates work for items that were already processed
const allItems = await getAllItems();
await Promise.all(allItems.map((item) => triggerExecute(item)));

// Good — keeps the pipeline idempotent
const allItems = await getAllItems();
const itemsNeedingProcessing = await filterExisting(allItems);
await Promise.all(itemsNeedingProcessing.map((item) => triggerExecute(item)));

❌ Inconsistent logging

// Bad — different prefixes, mixed formats
console.log('Starting workflow');
log.info('Processing item:', itemId);
console.log(`Done with ${itemId}`);

// Good — uniform prefix lets you grep by workflow+layer
console.log('[workflow:layer] Starting with payload:', payload);
console.log('[workflow:layer] Processing item:', { itemId });
console.log('[workflow:layer] Completed:', { itemId, result });

Cloud Project Workflow Configuration

Cloud-specific workflow configurations and patterns for the lobehub-cloud project.

Table of Contents

1. Overview 2. Directory Structure — submodule + cloud layout 3. Cloud-Specific Patterns — cloud-only workflows + re-export pattern 4. TypeScript Path Mappings 5. Workflow Class Location — cloud-only vs shared 6. Environment Variables 7. Best Practices — decide cloud vs OSS, re-export rules, naming 8. Migration Guide — moving workflows from cloud to lobehub 9. Examples — welcome-placeholder, agent-eval-run 10. Troubleshooting — circular imports, 404s, type errors 11. Related Documentation

Overview

The lobehub-cloud project extends the open-source lobehub codebase with cloud-specific features. Workflows can be implemented in either:

1. Lobehub (open-source) - Available to all users 2. Lobehub-cloud (proprietary) - Cloud-specific business logic

---

Directory Structure

Lobehub Submodule (Open-source)

lobehub/
└── src/
    ├── app/(backend)/api/workflows/
    │   ├── memory-user-memory/       # Memory extraction workflows
    │   └── agent-eval-run/            # Benchmark evaluation workflows
    └── server/workflows/
        ├── agentEvalRun/
        └── ...

Lobehub-cloud (Proprietary)

lobehub-cloud/
└── src/
    ├── app/(backend)/api/workflows/
    │   ├── welcome-placeholder/       # Cloud-only: AI placeholder generation
    │   ├── agent-welcome/            # Cloud-only: Agent welcome messages
    │   ├── agent-eval-run/           # Re-export from lobehub
    │   └── memory-user-memory/       # Re-export from lobehub
    └── server/workflows/
        ├── welcomePlaceholder/
        ├── agentWelcome/
        └── agentEvalRun/             # Re-export from lobehub

---

Cloud-Specific Patterns

Pattern 1: Cloud-Only Workflows

Use Case: Features exclusive to cloud users (AI generation, premium features)

Example: welcome-placeholder, agent-welcome

Implementation:

• Implement directly in lobehub-cloud/src/app/(backend)/api/workflows/
• No need for re-exports
• Can use cloud-specific packages and services

Structure:

lobehub-cloud/src/
├── app/(backend)/api/workflows/
│   └── feature-name/
│       ├── process-items/route.ts
│       ├── paginate-items/route.ts
│       └── execute-item/route.ts
└── server/workflows/
    └── featureName/
        └── index.ts

---

Pattern 2: Re-export from Lobehub

Use Case: Workflows implemented in open-source but also used in cloud

Example: agent-eval-run, memory-user-memory

Why Re-export?

• Cloud deployment needs to serve these endpoints
• Lobehub submodule code is not directly accessible in cloud routes
• Allows cloud-specific overrides if needed in the future

Re-export Implementation

Step 1: Implement workflow in lobehub submodule

// lobehub/src/app/(backend)/api/workflows/feature/layer/route.ts
import { serve } from '@upstash/workflow/nextjs';

export const { POST } = serve<Payload>(
  async (context) => {
    // Implementation
  },
  { flowControl: { ... } }
);

Step 2: Create re-export in lobehub-cloud

// lobehub-cloud/src/app/(backend)/api/workflows/feature/layer/route.ts
export { POST } from 'lobehub/src/app/(backend)/api/workflows/feature/layer/route';

Important: Use lobehub/src/... path, NOT @/... to avoid circular imports.

Re-export Directory Structure

# Create directories
mkdir -p lobehub-cloud/src/app/(backend)/api/workflows/feature-name/layer-1
mkdir -p lobehub-cloud/src/app/(backend)/api/workflows/feature-name/layer-2
mkdir -p lobehub-cloud/src/app/(backend)/api/workflows/feature-name/layer-3

# Create re-export files
echo "export { POST } from 'lobehub/src/app/(backend)/api/workflows/feature-name/layer-1/route';" > \
  lobehub-cloud/src/app/(backend)/api/workflows/feature-name/layer-1/route.ts

echo "export { POST } from 'lobehub/src/app/(backend)/api/workflows/feature-name/layer-2/route';" > \
  lobehub-cloud/src/app/(backend)/api/workflows/feature-name/layer-2/route.ts

echo "export { POST } from 'lobehub/src/app/(backend)/api/workflows/feature-name/layer-3/route';" > \
  lobehub-cloud/src/app/(backend)/api/workflows/feature-name/layer-3/route.ts

---

TypeScript Path Mappings

The cloud project uses tsconfig path mappings to override lobehub code:

// lobehub-cloud/tsconfig.json
{
  "compilerOptions": {
    "paths": {
      "@/*": ["./src/*", "./lobehub/src/*"]
    }
  }
}

Resolution Order:

1. ./src/* (cloud code) - checked first 2. ./lobehub/src/* (open-source) - fallback

This allows cloud to override specific modules while using lobehub defaults.

---

Workflow Class Location

Cloud-Only Workflows

Place workflow class in cloud:

lobehub-cloud/apps/server/src/workflows/featureName/index.ts

Shared Workflows

Place workflow class in lobehub, re-export in cloud if needed:

lobehub/apps/server/src/workflows/featureName/index.ts

---

Environment Variables

Both lobehub and cloud workflows require:

# Required for all workflows
APP_URL=https://your-app.com # Base URL for workflow endpoints
QSTASH_TOKEN=qstash_xxx      # QStash authentication token

# Optional (for custom QStash URL)
QSTASH_URL=https://custom-qstash.com # Custom QStash endpoint

Cloud-Specific:

# Cloud database (for monetization features)
CLOUD_DATABASE_URL=postgresql://...

# Cloud-specific services
REDIS_URL=redis://...

---

Best Practices

1. Decide: Cloud or Open-Source?

Implement in Lobehub if:

• Feature is useful for all LobeHub users
• No proprietary business logic
• Can be open-sourced

Implement in Cloud if:

• Premium/paid feature
• Uses cloud-specific services
• Contains proprietary algorithms

2. Re-export Pattern

✅ Do:

// Simple re-export
export { POST } from 'lobehub/src/app/(backend)/api/workflows/feature/route';

❌ Don't:

// Avoid circular imports with @/ path
export { POST } from '@/app/(backend)/api/workflows/feature/route'; // ❌

3. Keep Workflow Logic in Lobehub

For shared features:

• Implement core logic in lobehub/ (open-source)
• Only override if cloud needs different behavior
• Use re-exports for cloud deployment

4. Directory Naming

Follow consistent naming across lobehub and cloud:

# Both should use same structure
lobehub/src/app/(backend)/api/workflows/feature-name/
lobehub-cloud/src/app/(backend)/api/workflows/feature-name/

---

Migration Guide

Moving Workflow from Cloud to Lobehub

Step 1: Copy workflow to lobehub

cp -r lobehub-cloud/src/app/(backend)/api/workflows/feature \
      lobehub/src/app/(backend)/api/workflows/

Step 2: Remove cloud-specific dependencies

• Replace cloud services with generic interfaces
• Remove proprietary business logic
• Update imports to use lobehub paths

Step 3: Create re-exports in cloud

// lobehub-cloud/src/app/(backend)/api/workflows/feature/*/route.ts
export { POST } from 'lobehub/src/app/(backend)/api/workflows/feature/*/route';

Step 4: Move workflow class to lobehub

mv lobehub-cloud/apps/server/src/workflows/feature \
  lobehub/apps/server/src/workflows/

Step 5: Update cloud imports

// Change from
import { Workflow } from '@/server/workflows/feature';

// To
import { Workflow } from 'lobehub/apps/server/src/workflows/feature';

---

Examples

Cloud-Only Workflow: welcome-placeholder

Location: lobehub-cloud/src/app/(backend)/api/workflows/welcome-placeholder/

Why Cloud-Only: Uses proprietary AI generation service and Redis caching

Structure:

lobehub-cloud/
├── src/app/(backend)/api/workflows/welcome-placeholder/
│   ├── process-users/route.ts
│   ├── paginate-users/route.ts
│   └── generate-user/route.ts
└── apps/server/src/workflows/welcomePlaceholder/
    └── index.ts

Re-exported Workflow: agent-eval-run

Location:

• Implementation: lobehub/src/app/(backend)/api/workflows/agent-eval-run/
• Re-export: lobehub-cloud/src/app/(backend)/api/workflows/agent-eval-run/

Why Re-export: Core feature available in open-source, also used by cloud

Cloud Re-export Files:

// lobehub-cloud/src/app/(backend)/api/workflows/agent-eval-run/run-benchmark/route.ts
export { POST } from 'lobehub/src/app/(backend)/api/workflows/agent-eval-run/run-benchmark/route';

// lobehub-cloud/src/app/(backend)/api/workflows/agent-eval-run/paginate-test-cases/route.ts
export { POST } from 'lobehub/src/app/(backend)/api/workflows/agent-eval-run/paginate-test-cases/route';

// ... (all layers)

---

Troubleshooting

Circular Import Error

Error: Circular definition of import alias 'POST'

Cause: Using @/ path in re-export within cloud codebase

Solution: Use lobehub/src/ path instead

// ❌ Wrong
export { POST } from '@/app/(backend)/api/workflows/feature/route';

// ✅ Correct
export { POST } from 'lobehub/src/app/(backend)/api/workflows/feature/route';

Workflow Not Found (404)

Cause: Missing re-export in cloud

Solution: Create re-export files for all workflow layers

# Check if re-export exists
ls lobehub-cloud/src/app/\(backend\)/api/workflows/feature-name/

# If missing, create re-exports
mkdir -p lobehub-cloud/src/app/\(backend\)/api/workflows/feature-name/layer
echo "export { POST } from 'lobehub/src/app/(backend)/api/workflows/feature-name/layer/route';" > lobehub-cloud/src/app/\(backend\)/api/workflows/feature-name/layer/route.ts

Type Errors After Moving to Lobehub

Cause: Cloud-specific types or services used in lobehub code

Solution:

1. Extract cloud-specific logic to cloud-only wrapper 2. Use dependency injection for services 3. Define generic interfaces in lobehub

---

Related Documentation

• SKILL.md - Standard workflow patterns

Worked Examples

Two real workflows already in the codebase that follow this skill's pattern verbatim. Skim them when you want to see the pattern applied to concrete entities.

Example 1: Welcome Placeholder

Use case: Generate AI-powered welcome placeholders for users.

Structure:

• Layer 1: process-users — entry point, checks eligible users
• Layer 2: paginate-users — paginates through active users
• Layer 3: generate-user — generates placeholders for ONE user

Key features:

• Filters users who already have cached placeholders in Redis
• paidOnly flag to scope to subscribed users
• dryRun mode for statistics
• Fan-out for large user batches (CHUNK_SIZE=20)

Layer 3 shape:

export const { POST } = serve<GenerateUserPlaceholderPayload>(async (context) => {
  const { userId } = context.requestPayload ?? {};

  const workflow = new WelcomePlaceholderWorkflow(db, userId);
  const placeholders = await context.run('generate', () => workflow.generate());

  return { success: true, userId, placeholdersCount: placeholders.length };
});

Files:

• /api/workflows/welcome-placeholder/process-users/route.ts
• /api/workflows/welcome-placeholder/paginate-users/route.ts
• /api/workflows/welcome-placeholder/generate-user/route.ts
• /server/workflows/welcomePlaceholder/index.ts

---

Example 2: Agent Welcome

Use case: Generate welcome messages and open questions for AI agents.

Structure:

• Layer 1: process-agents — entry point, checks eligible agents
• Layer 2: paginate-agents — paginates through active agents
• Layer 3: generate-agent — generates welcome data for ONE agent

Key features:

• Filters agents who already have cached data in Redis
• paidOnly flag for subscribed users' agents only
• dryRun mode for statistics
• Fan-out for large agent batches (CHUNK_SIZE=20)

Layer 3 shape:

export const { POST } = serve<GenerateAgentWelcomePayload>(async (context) => {
  const { agentId } = context.requestPayload ?? {};

  const workflow = new AgentWelcomeWorkflow(db, agentId);
  const data = await context.run('generate', () => workflow.generate());

  return { success: true, agentId, data };
});

Files:

• /api/workflows/agent-welcome/process-agents/route.ts
• /api/workflows/agent-welcome/paginate-agents/route.ts
• /api/workflows/agent-welcome/generate-agent/route.ts
• /server/workflows/agentWelcome/index.ts

---

What's identical, what differs

Both workflows are the same pattern — they only differ in:

• Entity type (users vs agents)
• Business logic (placeholder generation vs welcome generation)
• Data source (different database queries)

Everything else — the 3-layer split, dry-run handling, fan-out, filter-existing, flowControl tuning — is identical. That's the whole point: once you internalize the pattern, adding a new workflow is mostly entity-substitution.

Implementation Patterns

Full code templates for the 3-layer architecture. Read this when actually writing workflow files.

Table of Contents

1. Workflow Class — apps/server/src/workflows/{workflowName}/index.ts 2. Layer 1: Entry Point — process-* route 3. Layer 2: Pagination — paginate-* route 4. Layer 3: Execution — execute-* / generate-* route

---

Workflow Class

Location: apps/server/src/workflows/{workflowName}/index.ts

import { Client } from '@upstash/workflow';
import debug from 'debug';

const log = debug('lobe-server:workflows:{workflow-name}');

// Workflow paths
const WORKFLOW_PATHS = {
  processItems: '/api/workflows/{workflow-name}/process-items',
  paginateItems: '/api/workflows/{workflow-name}/paginate-items',
  executeItem: '/api/workflows/{workflow-name}/execute-item',
} as const;

// Payload types
export interface ProcessItemsPayload {
  dryRun?: boolean;
  force?: boolean;
}

export interface PaginateItemsPayload {
  cursor?: string;
  itemIds?: string[]; // For fanout chunks
}

export interface ExecuteItemPayload {
  itemId: string;
}

const getWorkflowUrl = (path: string): string => {
  const baseUrl = process.env.APP_URL;
  if (!baseUrl) throw new Error('APP_URL is required to trigger workflows');
  return new URL(path, baseUrl).toString();
};

const getWorkflowClient = (): Client => {
  const token = process.env.QSTASH_TOKEN;
  if (!token) throw new Error('QSTASH_TOKEN is required to trigger workflows');

  const config: ConstructorParameters<typeof Client>[0] = { token };
  if (process.env.QSTASH_URL) {
    (config as Record<string, unknown>).url = process.env.QSTASH_URL;
  }
  return new Client(config);
};

export class {WorkflowName}Workflow {
  private static client: Client;

  private static getClient(): Client {
    if (!this.client) this.client = getWorkflowClient();
    return this.client;
  }

  static triggerProcessItems(payload: ProcessItemsPayload) {
    const url = getWorkflowUrl(WORKFLOW_PATHS.processItems);
    log('Triggering process-items workflow');
    return this.getClient().trigger({ body: payload, url });
  }

  static triggerPaginateItems(payload: PaginateItemsPayload) {
    const url = getWorkflowUrl(WORKFLOW_PATHS.paginateItems);
    log('Triggering paginate-items workflow');
    return this.getClient().trigger({ body: payload, url });
  }

  static triggerExecuteItem(payload: ExecuteItemPayload) {
    const url = getWorkflowUrl(WORKFLOW_PATHS.executeItem);
    log('Triggering execute-item workflow: %s', payload.itemId);
    return this.getClient().trigger({ body: payload, url });
  }

  /**
   * Filter items that need processing (e.g. check Redis cache, database state).
   * Return only the ones that actually need work — keeps the pipeline idempotent.
   */
  static async filterItemsNeedingProcessing(itemIds: string[]): Promise<string[]> {
    if (itemIds.length === 0) return [];
    // Check existing state and return items that need processing
    return itemIds;
  }
}

---

Layer 1: Entry Point (process-\*)

Purpose: Validates prerequisites, calculates statistics, supports dry-run mode.

import { serve } from '@upstash/workflow/nextjs';
import { getServerDB } from '@/database/server';
import { WorkflowClass, type ProcessPayload } from '@/server/workflows/{workflowName}';

export const { POST } = serve<ProcessPayload>(
  async (context) => {
    const { dryRun, force } = context.requestPayload ?? {};

    console.log('[{workflow}:process] Starting with payload:', { dryRun, force });

    const allItemIds = await context.run('{workflow}:get-all-items', async () => {
      const db = await getServerDB();
      // Query database for eligible items
      return items.map((item) => item.id);
    });

    console.log('[{workflow}:process] Total eligible items:', allItemIds.length);

    if (allItemIds.length === 0) {
      return { success: true, totalEligible: 0, message: 'No eligible items found' };
    }

    const itemsNeedingProcessing = await context.run('{workflow}:filter-existing', () =>
      WorkflowClass.filterItemsNeedingProcessing(allItemIds),
    );

    const result = {
      success: true,
      totalEligible: allItemIds.length,
      toProcess: itemsNeedingProcessing.length,
      alreadyProcessed: allItemIds.length - itemsNeedingProcessing.length,
    };

    // Dry-run short-circuits before any side effects
    if (dryRun) {
      console.log('[{workflow}:process] Dry run mode, returning statistics only');
      return {
        ...result,
        dryRun: true,
        message: `[DryRun] Would process ${itemsNeedingProcessing.length} items`,
      };
    }

    if (itemsNeedingProcessing.length === 0) {
      return { ...result, message: 'All items already processed' };
    }

    await context.run('{workflow}:trigger-paginate', () => WorkflowClass.triggerPaginateItems({}));

    return {
      ...result,
      message: `Triggered pagination for ${itemsNeedingProcessing.length} items`,
    };
  },
  {
    flowControl: {
      key: '{workflow}.process',
      parallelism: 1, // single instance — avoids duplicate processing
      ratePerSecond: 1,
    },
  },
);

---

Layer 2: Pagination (paginate-\*)

Purpose: Handles cursor-based pagination, implements fan-out for large batches.

import { serve } from '@upstash/workflow/nextjs';
import { chunk } from 'es-toolkit/compat';
import { getServerDB } from '@/database/server';
import { WorkflowClass, type PaginatePayload } from '@/server/workflows/{workflowName}';

const PAGE_SIZE = 50;
const CHUNK_SIZE = 20;

export const { POST } = serve<PaginatePayload>(
  async (context) => {
    const { cursor, itemIds: payloadItemIds } = context.requestPayload ?? {};

    console.log('[{workflow}:paginate] Starting:', {
      cursor,
      itemIdsCount: payloadItemIds?.length ?? 0,
    });

    // If specific itemIds were passed in (from a fanout chunk), process them directly
    if (payloadItemIds && payloadItemIds.length > 0) {
      await Promise.all(
        payloadItemIds.map((itemId) =>
          context.run(`{workflow}:execute:${itemId}`, () =>
            WorkflowClass.triggerExecuteItem({ itemId }),
          ),
        ),
      );
      return { success: true, processedItems: payloadItemIds.length };
    }

    // Paginate through all items
    const itemBatch = await context.run('{workflow}:get-batch', async () => {
      const db = await getServerDB();
      const items = await db.query(...);
      if (!items.length) return { ids: [] };
      const last = items.at(-1);
      return {
        ids: items.map((item) => item.id),
        cursor: last ? last.id : undefined,
      };
    });

    const batchItemIds = itemBatch.ids;
    const nextCursor = 'cursor' in itemBatch ? itemBatch.cursor : undefined;

    if (batchItemIds.length === 0) {
      return { success: true, message: 'Pagination complete' };
    }

    const itemIds = await context.run('{workflow}:filter-existing', () =>
      WorkflowClass.filterItemsNeedingProcessing(batchItemIds),
    );

    if (itemIds.length > 0) {
      if (itemIds.length > CHUNK_SIZE) {
        // Fan out — recursively re-enter pagination with each chunk
        const chunks = chunk(itemIds, CHUNK_SIZE);
        console.log('[{workflow}:paginate] Fanout mode:', {
          chunks: chunks.length,
          chunkSize: CHUNK_SIZE,
        });

        await Promise.all(
          chunks.map((ids, idx) =>
            context.run(`{workflow}:fanout:${idx + 1}/${chunks.length}`, () =>
              WorkflowClass.triggerPaginateItems({ itemIds: ids }),
            ),
          ),
        );
      } else {
        // Process this page directly
        await Promise.all(
          itemIds.map((itemId) =>
            context.run(`{workflow}:execute:${itemId}`, () =>
              WorkflowClass.triggerExecuteItem({ itemId }),
            ),
          ),
        );
      }
    }

    // Tail-call into the next page
    if (nextCursor) {
      await context.run('{workflow}:next-page', () =>
        WorkflowClass.triggerPaginateItems({ cursor: nextCursor }),
      );
    }

    return {
      success: true,
      processedItems: itemIds.length,
      skippedItems: batchItemIds.length - itemIds.length,
      nextCursor: nextCursor ?? null,
    };
  },
  {
    flowControl: {
      key: '{workflow}.paginate',
      parallelism: 20,
      ratePerSecond: 5,
    },
  },
);

---

Layer 3: Execution (execute-\ / generate-\)

Purpose: Performs the actual business logic for exactly ONE item.

import { serve } from '@upstash/workflow/nextjs';
import { getServerDB } from '@/database/server';
import { WorkflowClass, type ExecutePayload } from '@/server/workflows/{workflowName}';

export const { POST } = serve<ExecutePayload>(
  async (context) => {
    const { itemId } = context.requestPayload ?? {};

    if (!itemId) {
      return { success: false, error: 'Missing itemId' };
    }

    const db = await getServerDB();

    const item = await context.run('{workflow}:get-item', async () => {
      // Query database for item
      return item;
    });

    if (!item) {
      return { success: false, error: 'Item not found' };
    }

    const result = await context.run('{workflow}:process-item', async () => {
      const workflow = new WorkflowClass(db, itemId);
      return workflow.generate(); // or process(), execute(), etc.
    });

    await context.run('{workflow}:save-result', async () => {
      const workflow = new WorkflowClass(db, itemId);
      return workflow.saveToRedis(result); // or saveToDatabase(), etc.
    });

    return { success: true, itemId, result };
  },
  {
    flowControl: {
      key: '{workflow}.execute',
      parallelism: 10,
      ratePerSecond: 5,
    },
  },
);