Temporal.io + TypeScript SDK로 장기 실행 워크플로 만들기 — Saga·재시도·Signal을 코드로 선언하기

import { proxyActivities } from '@temporalio/workflow';
import type * as activities from './activities';
 
const {
  reserveInventory,
  processPayment,
  arrangeShipping,
  sendConfirmationEmail,
  updateOrderStatus,
} = proxyActivities<typeof activities>({
  startToCloseTimeout: '30s',
  retry: {
    initialInterval: '1s',
    backoffCoefficient: 2,
    maximumAttempts: 5,
    nonRetryableErrorTypes: ['CreditCardExpiredException', 'InvalidOrderException'],
  },
});
 
// 보상 Activity는 별도 프록시로 분리 — 재시도를 더 공격적으로
const {
  releaseInventory,
  refundPayment,
  cancelShipping,
} = proxyActivities<typeof activities>({
  startToCloseTimeout: '1m',
  retry: {
    initialInterval: '2s',
    backoffCoefficient: 2,
    maximumAttempts: 20,
  },
});

import { log } from '@temporalio/workflow';
 
export async function orderWorkflow(orderId: string): Promise<void> {
  const compensations: Array<() => Promise<void>> = [];
 
  try {
    await reserveInventory(orderId);
    compensations.push(() => releaseInventory(orderId));
 
    await processPayment(orderId);
    compensations.push(() => refundPayment(orderId));
 
    await arrangeShipping(orderId);
    compensations.push(() => cancelShipping(orderId));
 
    await sendConfirmationEmail(orderId);
  } catch (err) {
    for (const compensate of [...compensations].reverse()) {
      try {
        await compensate();
      } catch (compensationErr) {
        // 보상 실패는 로그로 남기되 나머지 보상은 계속 진행
        log.error('Compensation failed', { orderId, compensationErr });
      }
    }
    throw err;
  }
}

src/
  workflows/
    orderWorkflow.ts      ← 비즈니스 흐름 선언
  activities/
    inventoryActivities.ts
    paymentActivities.ts
    shippingActivities.ts
    notificationActivities.ts
    orderActivities.ts
  worker.ts               ← Worker 프로세스 진입점
  client.ts               ← 워크플로 시작

// activities/paymentActivities.ts
import { ApplicationFailure } from '@temporalio/activity';
 
export async function processPayment(orderId: string): Promise<void> {
  const result = await paymentGateway.charge(orderId);
 
  if (result.errorCode === 'CARD_EXPIRED') {
    throw ApplicationFailure.nonRetryable(
      'Card has expired',
      'CreditCardExpiredException',
    );
  }
 
  if (!result.success) {
    throw new Error(`Payment failed: ${result.message}`);
  }
}
 
export async function refundPayment(orderId: string): Promise<void> {
  await paymentGateway.refund(orderId);
}

// worker.ts
import { Worker } from '@temporalio/worker';
import * as inventoryActivities from './activities/inventoryActivities';
import * as paymentActivities from './activities/paymentActivities';
import * as shippingActivities from './activities/shippingActivities';
import * as notificationActivities from './activities/notificationActivities';
import * as orderActivities from './activities/orderActivities';
 
async function run() {
  // 주의: 여러 모듈을 spread하면 같은 이름의 export가 있을 때
  // 뒤에 오는 모듈이 앞을 조용히 덮어씁니다. 각 파일의 export 이름을
  // 겹치지 않게 관리하거나, 접두어 붙인 객체를 명시적으로 구성하세요.
  const worker = await Worker.create({
    workflowsPath: require.resolve('./workflows/orderWorkflow'),
    activities: {
      ...inventoryActivities,
      ...paymentActivities,
      ...shippingActivities,
      ...notificationActivities,
      ...orderActivities,
    },
    taskQueue: 'order-processing',
  });
 
  await worker.run();
}
 
run().catch(console.error);

// client.ts
import { Client, Connection } from '@temporalio/client';
 
async function startOrder(orderId: string) {
  const connection = await Connection.connect({ address: 'localhost:7233' });
  const client = new Client({ connection });
 
  const handle = await client.workflow.start('orderWorkflow', {
    taskQueue: 'order-processing',
    workflowId: `order-${orderId}`,
    args: [orderId],
  });
 
  console.log(`워크플로 시작: ${handle.workflowId}`);
 
  const result = await handle.result();
  return result;
}

import { defineSignal, setHandler, condition } from '@temporalio/workflow';
 
export const deliveryConfirmedSignal = defineSignal<[{ trackingCode: string }]>(
  'deliveryConfirmed',
);
 
export async function orderWorkflow(orderId: string): Promise<void> {
  // ... Saga 로직 ...
 
  let delivered = false;
  let trackingCode = '';
 
  setHandler(deliveryConfirmedSignal, ({ trackingCode: code }) => {
    delivered = true;
    trackingCode = code;
  });
 
  // condition은 조건이 참이 되면 true, 타임아웃이면 false를 반환합니다
  const arrived = await condition(() => delivered, '7d');
 
  if (arrived) {
    await updateOrderStatus(orderId, 'DELIVERED', trackingCode);
  } else {
    // 배송사 웹훅이 오지 않은 경우 — 별도 조사 큐로 넘기거나 재시도 워크플로 시작
    await updateOrderStatus(orderId, 'DELIVERY_TIMEOUT', '');
  }
}

await client.workflow
  .getHandle(`order-${orderId}`)
  .signal(deliveryConfirmedSignal, { trackingCode: 'KR1234567890' });

import { TestWorkflowEnvironment } from '@temporalio/testing';
import { Worker } from '@temporalio/worker';
 
let testEnv: TestWorkflowEnvironment;
 
beforeAll(async () => {
  testEnv = await TestWorkflowEnvironment.createLocal();
});
 
afterAll(async () => {
  await testEnv.teardown();
});
 
test('결제 실패 시 재고 예약이 취소된다', async () => {
  const { client, nativeConnection } = testEnv;
 
  const mockActivities = {
    reserveInventory: jest.fn().mockResolvedValue(undefined),
    processPayment: jest.fn().mockRejectedValue(new Error('Payment failed')),
    releaseInventory: jest.fn().mockResolvedValue(undefined),
    refundPayment: jest.fn().mockResolvedValue(undefined),
    arrangeShipping: jest.fn().mockResolvedValue(undefined),
    cancelShipping: jest.fn().mockResolvedValue(undefined),
    sendConfirmationEmail: jest.fn().mockResolvedValue(undefined),
    updateOrderStatus: jest.fn().mockResolvedValue(undefined),
  };
 
  const worker = await Worker.create({
    connection: nativeConnection,
    taskQueue: 'test-queue',
    workflowsPath: require.resolve('./workflows/orderWorkflow'),
    activities: mockActivities,
  });
 
  await worker.runUntil(async () => {
    // 워크플로가 최종적으로 실패하는 것이 이 테스트의 기대 동작.
    // 따라서 rejects.toThrow로 실패 자체를 검증하고, 이후 보상 호출 여부를 확인합니다.
    await expect(
      client.workflow.execute('orderWorkflow', {
        taskQueue: 'test-queue',
        workflowId: 'test-order-1',
        args: ['order-1'],
      }),
    ).rejects.toThrow(/Payment failed/);
  });
 
  expect(mockActivities.reserveInventory).toHaveBeenCalledWith('order-1');
  expect(mockActivities.releaseInventory).toHaveBeenCalledWith('order-1');
  expect(mockActivities.refundPayment).not.toHaveBeenCalled();
});

// 잘못된 방법: 워크플로 안에서 직접 HTTP 호출
export async function orderWorkflow(orderId: string) {
  const res = await fetch(`https://api.example.com/orders/${orderId}`); // 금지
}
 
// 올바른 방법: Activity로 감싸서 프록시로 호출
export async function orderWorkflow(orderId: string) {
  const order = await fetchOrderDetails(orderId); // proxyActivities로 노출된 함수
}

import { continueAsNew, workflowInfo } from '@temporalio/workflow';
 
export async function longRunningWorkflow(iteration: number): Promise<void> {
  // ... 작업 수행 ...
 
  // 상한의 절반쯤 도달하면 안전하게 새 인스턴스로 넘김
  if (workflowInfo().historyLength > 10_000) {
    await continueAsNew<typeof longRunningWorkflow>(iteration + 1);
  }
}