Workflows

Workflows allow you to automate business processes by chaining together actions, conditions, and integrations. You can trigger workflows via API, webhooks, or programmatically from your application code.

Triggering Workflows Programmatically

Use the triggerWorkflow function to start a workflow execution from anywhere in your server-side code, such as DataSource hooks, server actions, or background jobs.

Import

import { triggerWorkflow } from '@wayvo-ai/core/server';

Basic Usage

const result = await triggerWorkflow({
  workflowId: 'your-workflow-id',
  userName: 'user@example.com',
  input: { key: 'value' },
});

console.log('Execution started:', result.executionId);

Parameters

Return Value

interface TriggerWorkflowResult {
  executionId: string;  // UUID for tracking the execution
  status: 'running';    // Initial status
}

Using in DataSource Hooks

A common use case is triggering workflows when data changes. Use the afterInsert, afterUpdate, or afterDelete hooks to start a workflow in response to database operations.

Example: Trigger Workflow After Insert

import { triggerWorkflow } from '@wayvo-ai/core/server';
import type { DataSource } from '@wayvo-ai/core/common';

export const OrderDS: DataSource<Order> = {
  // ... other configuration ...

  afterInsert: async ({ rows, session, client }) => {
    for (const row of rows) {
      // Trigger the order processing workflow
      const result = await triggerWorkflow({
        workflowId: 'order-processing-workflow-id',
        userName: session.user.userName,
        input: {
          orderId: row.id,
          customerId: row.customerId,
          totalAmount: row.totalAmount,
        },
        client, // Reuse the existing transaction
      });

      console.log(`Started workflow for order ${row.id}:`, result.executionId);
    }

    return rows;
  },
};

Example: Trigger Workflow on Status Change

afterUpdate: async ({ rows, previousRows, session, client }) => {
  if (!previousRows) return rows;

  for (let i = 0; i < rows.length; i++) {
    const prev = previousRows[i];
    const row = rows[i];

    // Only trigger when status changes to 'approved'
    if (prev?.status !== row.status && row.status === 'approved') {
      await triggerWorkflow({
        workflowId: 'approval-notification-workflow-id',
        userName: session.user.userName,
        input: {
          recordId: row.id,
          approvedBy: session.user.userName,
          approvedAt: new Date().toISOString(),
        },
        client,
      });
    }
  }

  return rows;
},

Using in Server Actions

You can also trigger workflows from custom server actions:

'use server';

import { triggerWorkflow } from '@wayvo-ai/core/server';
import { auth } from '@/auth';

export async function startOnboardingWorkflow(userId: string) {
  const session = await auth();
  if (!session?.user) {
    throw new Error('Unauthorized');
  }

  const result = await triggerWorkflow({
    workflowId: 'user-onboarding-workflow-id',
    userName: session.user.userName,
    input: {
      userId,
      startedAt: new Date().toISOString(),
    },
  });

  return { executionId: result.executionId };
}

Error Handling

The triggerWorkflow function will throw an error if:

• The workflow is not found
• The workflow contains integration references that don't belong to the specified user

try {
  const result = await triggerWorkflow({
    workflowId: 'workflow-id',
    userName: session.user.userName,
    input: {},
  });
} catch (error) {
  if (error.message.includes('Workflow not found')) {
    // Handle missing workflow
  } else if (error.message.includes('invalid integration references')) {
    // Handle permission issue
  }
}

Tracking Execution Status

The returned executionId can be used to track the workflow execution status:

import { queryDataSource } from '@wayvo-ai/core/server';
import { getDataSource } from '@/lib/server/ds/defs';
import type { WorkflowExecutions } from '@/lib/common/ds/types/core/WorkflowExecutions';

// Query execution status
const executionsDS = getDataSource<WorkflowExecutions>('WorkflowExecutions');
const result = await queryDataSource(client, session, executionsDS, {
  filter: [{ id: { is: executionId } }],
});

const execution = result.rows[0];
console.log('Status:', execution?.status); // 'running' | 'success' | 'error' | 'paused'

Execution Lifecycle

1. running - Workflow is actively executing
2. paused - Workflow is waiting (e.g., sleep node)
3. success - Workflow completed successfully
4. error - Workflow failed after retries

Best Practices

Pass the Database Client

When calling from DataSource hooks, always pass the client parameter to reuse the existing transaction:

await triggerWorkflow({
  workflowId: 'workflow-id',
  userName: session.user.userName,
  input: { /* ... */ },
  client, // Important: reuses the hook's transaction
});

Use Meaningful Input Data

Pass relevant data in the input object that your workflow nodes will need:

input: {
  // Include IDs for lookups
  recordId: row.id,
  userId: session.user.userName,

  // Include data to avoid extra queries in the workflow
  customerEmail: row.email,
  orderTotal: row.totalAmount,

  // Include context
  triggeredAt: new Date().toISOString(),
  triggeredBy: session.user.displayName,
}

Handle Errors Gracefully

Workflow execution happens asynchronously. The triggerWorkflow function returns immediately after creating the execution record. Handle setup errors but don't expect workflow completion errors:

try {
  const result = await triggerWorkflow({ /* ... */ });
  // Execution started - workflow runs in background
  return { success: true, executionId: result.executionId };
} catch (error) {
  // Setup failed (workflow not found, permission error, etc.)
  console.error('Failed to start workflow:', error);
  return { success: false, error: error.message };
}

Custom Workflow Nodes

You can register custom workflow actions that appear in the workflow builder and can be executed by the workflow engine. Custom plugins are registered via configuration - no side-effect imports needed.

Plugin Structure

Custom workflow plugins have two parts:

1. Server Plugin (WorkflowServerPlugin) - Step functions and labels for execution
2. Client Plugin (WorkflowClientPlugin) - Integration definition for the workflow builder UI

Step 1: Create the Step Functions

Create step functions that implement your custom logic:

// src/lib/workflow/plugins/my-integration/steps.ts
import 'server-only';
import type { StepFunction } from '@wayvo-ai/core/server';

export const myCustomStep: StepFunction = async (input) => {
  const { message, userId } = input;

  // Your custom logic here
  const result = await doSomething(message, userId);

  return {
    success: true,
    data: result,
  };
};

Step 2: Create the Server Plugin

Define the server plugin with step importers and action labels:

// src/lib/workflow/plugins/my-integration/server.ts
import 'server-only';
import type { WorkflowServerPlugin } from '@wayvo-ai/core/server';
import { myCustomStep } from './steps';

export const myServerPlugin: WorkflowServerPlugin = {
  stepImporters: [
    { actionId: 'my-integration/my-action', importer: { stepFunction: myCustomStep } },
  ],
  actionLabels: [
    { actionId: 'my-integration/my-action', label: 'My Custom Action' },
  ],
};

Step 3: Create the Client Plugin

Define the client plugin with the integration for the workflow builder UI:

// src/lib/workflow/plugins/my-integration/client.tsx
import type { WorkflowClientPlugin } from '@wayvo-ai/core/ui';
import { MyIcon } from './icon';

export const myClientPlugin: WorkflowClientPlugin = {
  integration: {
    type: 'my-integration',
    label: 'My Integration',
    description: 'Custom integration for my app',
    icon: MyIcon,
    formFields: [], // No credentials needed for this example
    actions: [
      {
        slug: 'my-action',
        label: 'My Custom Action',
        description: 'Does something custom',
        category: 'My Integration', // Must match the integration label
        stepFunction: 'myCustomStep',
        stepImportPath: 'my-integration',
        configFields: [
          {
            key: 'message',
            label: 'Message',
            type: 'template-input',
            placeholder: 'Enter a message',
            required: true,
          },
        ],
        outputFields: [
          { field: 'success', description: 'Whether the action succeeded' },
          { field: 'data', description: 'The result data' },
        ],
      },
    ],
  },
};

Important: The category field in each action must match the integration's label for the category dropdown to work correctly.

Step 4: Register via Configuration

Add the plugins to your ServerConfig and AppProvider:

// src/lib/server/init/server-config.ts
import type { ServerConfig } from '@wayvo-ai/core/types';
import { myServerPlugin } from '@/lib/workflow/plugins/my-integration/server';

const serverConfig: ServerConfig = {
  // ... other config
  workflowPlugins: [myServerPlugin],
};

export default serverConfig;

// src/app/(secure)/layout.tsx
import { AppProvider } from '@wayvo-ai/core/ui';
import { myClientPlugin } from '@/lib/workflow/plugins/my-integration/client';

export default function Layout({ children }) {
  return (
    <AppProvider
      // ... other props
      workflowPlugins={[myClientPlugin]}
    >
      {children}
    </AppProvider>
  );
}

Complete Example

Here's a complete example of a custom "SMS" integration:

Step Functions

// src/lib/workflow/plugins/sms/steps.ts
import 'server-only';
import type { StepFunction } from '@wayvo-ai/core/server';

export const sendSmsStep: StepFunction = async (input) => {
  const { phoneNumber, message } = input;

  const response = await fetch('https://api.smsprovider.com/send', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ to: phoneNumber, body: message }),
  });

  if (!response.ok) {
    return { success: false, error: 'Failed to send SMS' };
  }

  const data = await response.json();
  return { success: true, messageId: data.id };
};

Server Plugin

// src/lib/workflow/plugins/sms/server.ts
import 'server-only';
import type { WorkflowServerPlugin } from '@wayvo-ai/core/server';
import { sendSmsStep } from './steps';

export const smsServerPlugin: WorkflowServerPlugin = {
  stepImporters: [
    { actionId: 'sms/send', importer: { stepFunction: sendSmsStep } },
  ],
  actionLabels: [
    { actionId: 'sms/send', label: 'Send SMS' },
  ],
};

Client Plugin

// src/lib/workflow/plugins/sms/client.tsx
import type { WorkflowClientPlugin } from '@wayvo-ai/core/ui';
import { MessageSquare } from 'lucide-react';

export const smsClientPlugin: WorkflowClientPlugin = {
  integration: {
    type: 'sms',
    label: 'SMS',
    description: 'Send SMS messages',
    icon: MessageSquare,
    formFields: [
      {
        id: 'apiKey',
        label: 'API Key',
        type: 'password',
        placeholder: 'Your SMS provider API key',
        configKey: 'apiKey',
      },
    ],
    actions: [
      {
        slug: 'send',
        label: 'Send SMS',
        description: 'Send an SMS message to a phone number',
        category: 'SMS', // Matches integration label
        stepFunction: 'sendSmsStep',
        stepImportPath: 'sms',
        configFields: [
          {
            key: 'phoneNumber',
            label: 'Phone Number',
            type: 'template-input',
            placeholder: '+1234567890',
            required: true,
          },
          {
            key: 'message',
            label: 'Message',
            type: 'template-textarea',
            placeholder: 'Your message here',
            rows: 3,
            required: true,
          },
        ],
        outputFields: [
          { field: 'success', description: 'Whether the SMS was sent' },
          { field: 'messageId', description: 'The SMS message ID' },
        ],
      },
    ],
  },
};

Dynamic Import for Code Splitting

For larger step implementations, use dynamic imports to reduce bundle size:

export const myServerPlugin: WorkflowServerPlugin = {
  stepImporters: [
    {
      actionId: 'my-integration/heavy-action',
      importer: {
        importer: () => import('./steps/heavy-action'),
        stepFunction: 'heavyActionStep', // Name of the exported function
      },
    },
  ],
};

Config Field Types

Available field types for configFields:

Combobox

Searchable dropdown. Options can be static, from a built-in source, or from a callback:

// Static options
{ key: 'priority', label: 'Priority', type: 'combobox', options: [{ value: 'high', label: 'High' }, ...] }

// Built-in source (e.g. datasources)
{ key: 'datasourceId', label: 'DataSource', type: 'combobox', optionsSource: 'datasources' }

// Dynamic options from config (e.g. depends on another field)
{
  key: 'attribute',
  label: 'Attribute',
  type: 'combobox',
  getOptions: async (config) => {
    const id = (config?.datasourceId as string) || '';
    if (!id) return [];
    const attrs = await getDataSourceAttributes(id);
    return attrs.map((a) => ({ value: a.code, label: a.name }));
  },
}

Optional: placeholder, searchPlaceholder.

Single-row-update

Single-row form for DataSource update, insert, or delete. Renders primary key fields first (from the selected DataSource), then optional “Fields to update” (attribute + template value pairs). Use singleRowMode to control behavior:

Requires a DataSource to be selected (e.g. via a datasourceId combobox) so attributes can be loaded.

// Update: one row object
{ key: 'row', label: 'Record to update', type: 'single-row-update', required: true }

// Insert: one row serialized as [row]
{ key: 'rows', label: 'Record to insert', type: 'single-row-update', singleRowMode: 'insert', required: true }

// Delete: identify record by primary key(s), serialized as [row]
{ key: 'rows', label: 'Record to delete', type: 'single-row-update', singleRowMode: 'delete', required: true }

Match-builder

Dynamic list of attribute + value pairs, stored as a single JSON object (e.g. for DataSource query match). Options for the attribute dropdown can come from getOptions(config) (e.g. DataSource attributes). Values support template syntax.

Multi-select

Multi-select that stores the selection as a JSON array string. Options from options or getOptions(config). Use optionsDependencyKeys so options are refetched only when those config values change (e.g. avoid refetch when the field’s own value changes):

{
  key: 'select',
  label: 'Select fields',
  type: 'multi-select',
  placeholder: 'Select fields to return (empty = all)',
  searchPlaceholder: 'Search attributes...',
  optionsDependencyKeys: ['datasourceId'],
  getOptions: async (config) => {
    const id = (config?.datasourceId as string) || '';
    if (!id) return [];
    const attrs = await getDataSourceAttributes(id);
    return attrs.map((a) => ({ value: a.code, label: a.name }));
  },
}

Optional: placeholder, searchPlaceholder.

Conditional Fields

Show fields conditionally based on other field values:

configFields: [
  {
    key: 'notificationType',
    label: 'Type',
    type: 'select',
    options: [
      { value: 'email', label: 'Email' },
      { value: 'sms', label: 'SMS' },
    ],
  },
  {
    key: 'emailAddress',
    label: 'Email Address',
    type: 'template-input',
    showWhen: { field: 'notificationType', equals: 'email' },
  },
  {
    key: 'phoneNumber',
    label: 'Phone Number',
    type: 'template-input',
    showWhen: { field: 'notificationType', equals: 'sms' },
  },
],

Displaying Workflow Execution Status

Wayvo Core provides React components to display workflow execution status in your application UI. These components show a visual canvas of the workflow with real-time status updates for each node.

Components

Import

import {
  WorkflowExecutionViewer,
  WorkflowExecutionViewerDialog,
  ReactFlowProvider,
} from '@wayvo-ai/core/ui';

WorkflowExecutionViewer

The WorkflowExecutionViewer component displays the workflow execution with:

• Visual canvas showing workflow nodes with their execution status
• Real-time updates via SSE for running executions
• Clickable nodes to view execution details and logs
• Timeline panel showing execution progress

Important: This component must be wrapped in a ReactFlowProvider.

Props

Basic Usage

'use client';

import { ReactFlowProvider, WorkflowExecutionViewer } from '@wayvo-ai/core/ui';

export function ExecutionStatus({ executionId, workflowId }: {
  executionId: string;
  workflowId: string;
}) {
  return (
    <ReactFlowProvider>
      <WorkflowExecutionViewer
        executionId={executionId}
        workflowId={workflowId}
        height="500px"
      />
    </ReactFlowProvider>
  );
}

Embedding in a Page

'use client';

import { ReactFlowProvider, WorkflowExecutionViewer } from '@wayvo-ai/core/ui';

export function OrderExecutionPage({ orderId, executionId, workflowId }: {
  orderId: string;
  executionId: string;
  workflowId: string;
}) {
  return (
    <div className="flex h-screen flex-col">
      <header className="border-b p-4">
        <h1>Order #{orderId} - Workflow Execution</h1>
      </header>
      <div className="flex-1">
        <ReactFlowProvider>
          <WorkflowExecutionViewer
            executionId={executionId}
            workflowId={workflowId}
            className="h-full w-full"
          />
        </ReactFlowProvider>
      </div>
    </div>
  );
}

WorkflowExecutionViewerDialog

The WorkflowExecutionViewerDialog component shows the execution viewer in a popup dialog. It handles the ReactFlowProvider internally.

Props

Usage

'use client';

import { useState } from 'react';
import { WorkflowExecutionViewerDialog } from '@wayvo-ai/core/ui';
import { Button } from '@/components/ui/button';

export function ViewExecutionButton({ executionId, workflowId }: {
  executionId: string;
  workflowId: string;
}) {
  const [isOpen, setIsOpen] = useState(false);

  return (
    <>
      <Button onClick={() => setIsOpen(true)}>
        View Execution
      </Button>
      {isOpen && (
        <WorkflowExecutionViewerDialog
          executionId={executionId}
          workflowId={workflowId}
          onClose={() => setIsOpen(false)}
        />
      )}
    </>
  );
}

Complete Example: Triggering and Viewing a Workflow

This example shows how to trigger a workflow and display its execution status:

'use client';

import { useState, useTransition } from 'react';
import { ReactFlowProvider, WorkflowExecutionViewer } from '@wayvo-ai/core/ui';
import { Button } from '@/components/ui/button';
import { startOrderWorkflow } from './actions';

export function OrderWorkflowPanel({ orderId }: { orderId: string }) {
  const [executionId, setExecutionId] = useState<string | null>(null);
  const [isPending, startTransition] = useTransition();

  const handleTrigger = () => {
    startTransition(async () => {
      const result = await startOrderWorkflow(orderId);
      if (result.executionId) {
        setExecutionId(result.executionId);
      }
    });
  };

  if (!executionId) {
    return (
      <Button onClick={handleTrigger} disabled={isPending}>
        {isPending ? 'Starting...' : 'Start Order Processing'}
      </Button>
    );
  }

  return (
    <div className="h-[600px]">
      <ReactFlowProvider>
        <WorkflowExecutionViewer
          executionId={executionId}
          workflowId="your-order-workflow-id"
          className="h-full w-full"
        />
      </ReactFlowProvider>
    </div>
  );
}

Server action:

// actions.ts
'use server';

import { triggerWorkflow } from '@wayvo-ai/core/server';
import { auth } from '@/auth';

export async function startOrderWorkflow(orderId: string) {
  const session = await auth();
  if (!session?.user) throw new Error('Unauthorized');

  const result = await triggerWorkflow({
    workflowId: 'your-order-workflow-id',
    userName: session.user.userName,
    input: { orderId },
  });

  return { executionId: result.executionId };
}

Node Status Colors

The viewer displays nodes with different colors based on their execution status:

Real-Time Updates

When viewing a running or paused execution, the components automatically subscribe to SSE updates. Node statuses, logs, and the timeline update in real-time as the workflow progresses.

Next Steps

• DataSources - Learn about lifecycle hooks
• Server Actions - Create custom server-side logic