Розробка Workflow-рушія на базі Temporal

Розроблення Workflow-рушія на базі Temporal

Temporal — платформа для надійного виконання довгострокових бізнес-процесів. Workflow-функції в Temporal виглядають як звичайний код, але Temporal автоматично зберігає їх стан, забезпечує повторне спробування при збоях та дозволяє workflow пережити рестарти серверів.

Чим Temporal відрізняється від черг повідомлень

У чергах повідомлень (RabbitMQ, Kafka) при збої між кроками потрібно самостійно керувати станом, dead letter queue та retry-логікою. У Temporal workflow-функція "засинає" та "прокидається" — рушій гарантує, що вона виконається до кінця.

Ключові концепції

Workflow — детерміністична функція, яка визначає порядок кроків. Може "спати" годинами/днями та чекати сигналів.

Activity — окремий крок з побічними ефектами (HTTP-запит, запис у БД). Activities мають retry-політику.

Worker — процес, який виконує Workflow та Activity код.

Signal — зовнішня подія, яка може змінити стан workflow (наприклад, "платіж підтверджений").

Query — читання поточного стану workflow без змін.

Встановлення 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';

// Визначення інтерфейсу activities
const { validateOrder, reserveInventory, processPayment,
        sendConfirmation, releaseInventory, refundPayment } =
  proxyActivities<typeof import('./activities')>({
    startToCloseTimeout: '30 seconds',
    retry: {
      maximumAttempts: 3,
      initialInterval: '1 second',
      backoffCoefficient: 2,
    }
  });

// Signals та 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}`;
  });

  // Крок 1: Валідація
  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' };

  // Крок 2: Резервування товара
  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' };
  }

  // Крок 3: Очікування підтвердження оплати (до 30 хвилин)
  status = 'awaiting_payment';
  let paymentId: string | null = null;

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

  // Чекаємо сигнал або таймаут
  await sleep('30 minutes');

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

  // Крок 4: Обробка платежу
  status = 'processing_payment';
  try {
    await processPayment(orderId, paymentId);
  } catch (e) {
    await releaseInventory(orderId);
    return { success: false, reason: 'Payment failed' };
  }

  // Крок 5: Підтвердження
  status = 'completed';
  await sendConfirmation(orderId);

  return { success: true, orderId };
}

Activities — реальні операції

// activities.ts
export async function validateOrder(orderId: string): Promise<ValidationResult> {
  // Реальний HTTP-запит або виклик БД тут
  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();

Запуск Workflow та відправлення сигналів

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

const client = new Client();

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

// Відправлення сигналу при підтвердженні оплати (з webhook Stripe)
await client.workflow.getHandle(`order-${orderId}`)
  .signal(paymentConfirmedSignal, { paymentId: stripePaymentId });

// Запит поточного статусу
const status = await client.workflow.getHandle(`order-${orderId}`)
  .query(orderStatusQuery);

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

Versioning

При змінюванні workflow-коду потрібна версіонізація, щоб не зламати активні екземпляри:

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

export async function orderWorkflow(orderId: string) {
  // Старі екземпляри використовують стару переходу
  // Нові — нову
  if (patched('add-fraud-check')) {
    await fraudCheck(orderId);
  }

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

Строки виконання

• Temporal Server (Docker) + один workflow з 3–5 activities — 1–2 тижні
• Складний workflow з сигналами, таймерами та versioning — 2–4 тижні
• Міграція існуючого saga/queue-based коду на Temporal — 1–2 місяці