Workflow Engine Development Based on Temporal

Developing Workflow Engine with Temporal

Temporal is a platform for reliable execution of long-running business processes. Workflow functions in Temporal look like regular code, but Temporal automatically saves their state, ensures retry on failures, and allows workflows to survive server restarts.

How Temporal Differs from Message Queues

In message queues (RabbitMQ, Kafka), you must manage state, dead letter queue, and retry logic yourself when failures occur between steps. In Temporal, a workflow function "sleeps" and "wakes up" — the engine guarantees it will execute to completion.

Key Concepts

Workflow — a deterministic function that defines the order of steps. Can "sleep" for hours/days and wait for signals.

Activity — a separate step with side effects (HTTP request, database write). Activities have retry policies.

Worker — a process that executes Workflow and Activity code.

Signal — an external event that can change workflow state (e.g., "payment confirmed").

Query — reading current workflow state without changes.

Setting Up Temporal Server

# docker-compose.yml
services:
  temporal:
    image: temporalio/auto-setup:1.22
    ports:
      - "7233:7233"
    environment:
      - DB=postgresql
      - DB_PORT=5432
      - POSTGRES_USER=temporal
      - POSTGRES_PWD=temporal
      - POSTGRES_SEEDS=postgresql
    depends_on:
      - postgresql

  temporal-ui:
    image: temporalio/ui:2.22
    ports:
      - "8080:8080"
    environment:
      - TEMPORAL_ADDRESS=temporal:7233

  postgresql:
    image: postgres:15-alpine
    environment:
      POSTGRES_USER: temporal
      POSTGRES_PASSWORD: temporal
      POSTGRES_DB: temporal

Workflow — Node.js SDK

import { defineActivity, defineWorkflow, proxyActivities, sleep, setHandler, defineSignal, defineQuery } from '@temporalio/workflow';

// Define activities interface
const { validateOrder, reserveInventory, processPayment,
        sendConfirmation, releaseInventory, refundPayment } =
  proxyActivities<typeof import('./activities')>({
    startToCloseTimeout: '30 seconds',
    retry: {
      maximumAttempts: 3,
      initialInterval: '1 second',
      backoffCoefficient: 2,
    }
  });

// Signals and Queries
const paymentConfirmedSignal = defineSignal<[{ paymentId: string }]>('paymentConfirmed');
const cancelOrderSignal = defineSignal<[{ reason: string }]>('cancelOrder');
const orderStatusQuery = defineQuery<string>('orderStatus');

// Workflow
export async function orderWorkflow(orderId: string): Promise<OrderResult> {
  let status = 'validating';
  let cancelled = false;

  setHandler(orderStatusQuery, () => status);

  setHandler(cancelOrderSignal, ({ reason }) => {
    cancelled = true;
    status = `cancelled: ${reason}`;
  });

  // Step 1: Validation
  status = 'validating';
  const validation = await validateOrder(orderId);
  if (!validation.valid) {
    return { success: false, reason: validation.reason };
  }

  if (cancelled) return { success: false, reason: 'Cancelled before reservation' };

  // Step 2: Inventory reservation
  status = 'reserving';
  let inventoryReserved = false;

  try {
    await reserveInventory(orderId, validation.items);
    inventoryReserved = true;
  } catch (e) {
    return { success: false, reason: 'Insufficient stock' };
  }

  if (cancelled) {
    await releaseInventory(orderId);
    return { success: false, reason: 'Cancelled' };
  }

  // Step 3: Wait for payment confirmation (up to 30 minutes)
  status = 'awaiting_payment';
  let paymentId: string | null = null;

  setHandler(paymentConfirmedSignal, ({ paymentId: pid }) => {
    paymentId = pid;
  });

  // Wait for signal or timeout
  await sleep('30 minutes');

  if (!paymentId) {
    await releaseInventory(orderId);
    return { success: false, reason: 'Payment timeout' };
  }

  // Step 4: Process payment
  status = 'processing_payment';
  try {
    await processPayment(orderId, paymentId);
  } catch (e) {
    await releaseInventory(orderId);
    return { success: false, reason: 'Payment failed' };
  }

  // Step 5: Confirmation
  status = 'completed';
  await sendConfirmation(orderId);

  return { success: true, orderId };
}

Activities — Real Operations

// activities.ts
export async function validateOrder(orderId: string): Promise<ValidationResult> {
  // Real HTTP request or DB call here
  const order = await orderRepository.findById(orderId);
  if (!order) throw new ApplicationFailure(`Order ${orderId} not found`);

  const itemsValid = await checkItemsAvailability(order.items);
  return { valid: itemsValid, items: order.items, reason: itemsValid ? null : 'Items unavailable' };
}

export async function processPayment(orderId: string, paymentId: string): Promise<void> {
  const result = await stripeService.capturePayment(paymentId);
  if (result.status !== 'succeeded') {
    throw new ApplicationFailure(`Payment capture failed: ${result.failureMessage}`);
  }
  await orderRepository.markAsPaid(orderId, paymentId);
}

Worker

// worker.ts
import { Worker } from '@temporalio/worker';
import * as activities from './activities';

const worker = await Worker.create({
  workflowsPath: require.resolve('./workflows'),
  activities,
  taskQueue: 'orders',
  maxConcurrentActivityTaskExecutions: 50,
  maxConcurrentWorkflowTaskExecutions: 50,
});

await worker.run();

Starting Workflow and Sending Signals

// client.ts
import { Client } from '@temporalio/client';

const client = new Client();

// Start workflow
const handle = await client.workflow.start(orderWorkflow, {
  taskQueue: 'orders',
  workflowId: `order-${orderId}`,
  args: [orderId],
});

// Send signal on payment confirmation (from Stripe webhook)
await client.workflow.getHandle(`order-${orderId}`)
  .signal(paymentConfirmedSignal, { paymentId: stripePaymentId });

// Query current status
const status = await client.workflow.getHandle(`order-${orderId}`)
  .query(orderStatusQuery);

console.log('Order status:', status); // 'awaiting_payment'

Versioning

When changing workflow code, versioning is needed to avoid breaking active instances:

import { patched } from '@temporalio/workflow';

export async function orderWorkflow(orderId: string) {
  // Old instances use old path
  // New instances use new path
  if (patched('add-fraud-check')) {
    await fraudCheck(orderId);
  }

  await validateOrder(orderId);
  // ...
}

Implementation Timeframe

• Temporal Server (Docker) + one workflow with 3–5 activities — 1–2 weeks
• Complex workflow with signals, timers, and versioning — 2–4 weeks
• Migration of existing saga/queue-based code to Temporal — 1–2 months