Temporal.io 워크플로우 튜토리얼 — TypeScript로 분산 작업 큐와 장기 실행 프로세스 구현

Temporal CLI 설치 및 개발 서버 실행
# macOS (Homebrew)
brew install temporal

# 개발 서버 시작 (localhost:7233, UI: localhost:8233)
temporal server start-dev

# 프로젝트 초기화
mkdir temporal-demo && cd temporal-demo
npm init -y
npm install @temporalio/workflow @temporalio/activity @temporalio/worker @temporalio/client
npm install -D typescript ts-node @types/node

# tsconfig.json
npx tsc --init --target es2020 --module commonjs

src/workflows.ts — 주문 처리 워크플로우
import { proxyActivities, sleep } from '@temporalio/workflow';
import type * as activities from './activities';

const { chargePayment, sendConfirmationEmail, updateOrderStatus } =
  proxyActivities<typeof activities>({
    startToCloseTimeout: '30 seconds',
    retry: {
      maximumAttempts: 3,
    },
  });

export interface OrderInput {
  orderId: string;
  userId: string;
  amount: number;
  email: string;
}

export async function orderWorkflow(input: OrderInput): Promise<string> {
  await updateOrderStatus(input.orderId, 'processing');

  // 결제 처리
  const paymentId = await chargePayment(input.userId, input.amount);

  await updateOrderStatus(input.orderId, 'paid');

  // 확인 이메일 발송
  await sendConfirmationEmail(input.email, input.orderId, paymentId);

  await updateOrderStatus(input.orderId, 'completed');

  return paymentId;
}

src/activities.ts — Activity 구현
import { Context } from '@temporalio/activity';

export async function chargePayment(
  userId: string,
  amount: number
): Promise<string> {
  // Activity는 I/O가 가능하다
  Context.current().heartbeat('charging...');

  // 실제 결제 API 호출 예시
  const response = await fetch('https://api.payment.example/charge', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ userId, amount }),
  });

  if (!response.ok) {
    throw new Error(`Payment failed: ${response.status}`);
  }

  const { paymentId } = await response.json();
  return paymentId;
}

export async function sendConfirmationEmail(
  email: string,
  orderId: string,
  paymentId: string
): Promise<void> {
  console.log(`Sending confirmation to ${email} for order ${orderId}`);
  // 이메일 전송 로직
}

export async function updateOrderStatus(
  orderId: string,
  status: string
): Promise<void> {
  console.log(`Order ${orderId} status → ${status}`);
  // DB 업데이트 로직
}

src/worker.ts — 워커 등록 및 실행
import { Worker } from '@temporalio/worker';
import * as activities from './activities';

async function main() {
  const worker = await Worker.create({
    workflowsPath: require.resolve('./workflows'),
    activities,
    taskQueue: 'order-processing',
    // 동시 실행 제한
    maxConcurrentActivityTaskExecutions: 10,
    maxConcurrentWorkflowTaskExecutions: 20,
  });

  console.log('Worker started, polling order-processing queue...');
  await worker.run();
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

src/client.ts — 워크플로우 시작 및 결과 대기
import { Client, Connection } from '@temporalio/client';
import { orderWorkflow } from './workflows';

async function main() {
  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-${Date.now()}`,
    args: [{
      orderId: 'order-001',
      userId: 'user-123',
      amount: 99000,
      email: 'user@example.com',
    }],
  });

  console.log(`Workflow started: ${handle.workflowId}`);

  // 완료될 때까지 대기
  const paymentId = await handle.result();
  console.log(`Completed, paymentId: ${paymentId}`);

  await connection.close();
}

main().catch(console.error);

Activity별 세밀한 재시도·타임아웃 설정
import { proxyActivities, ApplicationFailure } from '@temporalio/workflow';
import type * as activities from './activities';

// 결제: 재시도 3회, 타임아웃 30초
const { chargePayment } = proxyActivities<typeof activities>({
  startToCloseTimeout: '30 seconds',
  retry: {
    maximumAttempts: 3,
    initialInterval: '2 seconds',
    backoffCoefficient: 2.0,
    maximumInterval: '30 seconds',
    // 이 에러는 재시도하지 않음
    nonRetryableErrorTypes: ['PaymentDeclinedError'],
  },
});

// 이메일: 재시도 최대 10회, 최대 5분 간격
const { sendConfirmationEmail } = proxyActivities<typeof activities>({
  startToCloseTimeout: '60 seconds',
  scheduleToCloseTimeout: '10 minutes',
  retry: {
    maximumAttempts: 10,
    initialInterval: '5 seconds',
    backoffCoefficient: 1.5,
    maximumInterval: '5 minutes',
  },
});

// Activity 내에서 재시도 불가 에러 던지기
export async function chargePaymentImpl(userId: string, amount: number) {
  const result = await paymentGateway.charge(userId, amount);
  if (result.errorCode === 'CARD_DECLINED') {
    // nonRetryable = true → 재시도 안 함
    throw ApplicationFailure.create({
      type: 'PaymentDeclinedError',
      nonRetryable: true,
      message: '카드가 거절되었습니다',
    });
  }
}

Signal과 Query를 활용한 취소 가능한 워크플로우
import {
  proxyActivities,
  defineSignal,
  defineQuery,
  setHandler,
  condition,
} from '@temporalio/workflow';
import type * as activities from './activities';

const { chargePayment, refundPayment, updateOrderStatus } =
  proxyActivities<typeof activities>({ startToCloseTimeout: '30 seconds' });

// Signal/Query 타입 정의
export const cancelSignal = defineSignal<[{ reason: string }]>('cancel');
export const statusQuery = defineQuery<string>('getStatus');

export async function orderWorkflow(input: OrderInput): Promise<string> {
  let cancelled = false;
  let cancelReason = '';
  let currentStatus = 'started';

  // Signal 핸들러 등록
  setHandler(cancelSignal, ({ reason }) => {
    cancelled = true;
    cancelReason = reason;
  });

  // Query 핸들러 등록
  setHandler(statusQuery, () => currentStatus);

  currentStatus = 'charging';
  const paymentId = await chargePayment(input.userId, input.amount);

  // 취소 시그널이 왔으면 환불 후 종료
  if (cancelled) {
    currentStatus = 'refunding';
    await refundPayment(paymentId);
    await updateOrderStatus(input.orderId, 'cancelled');
    return `cancelled: ${cancelReason}`;
  }

  currentStatus = 'completed';
  return paymentId;
}

클라이언트에서 Signal 전송 및 Query 조회
import { Client, Connection } from '@temporalio/client';
import { orderWorkflow, cancelSignal, statusQuery } from './workflows';

const client = new Client({ connection: await Connection.connect() });

const handle = client.workflow.getHandle('order-001');

// 현재 상태 조회 (Query)
const status = await handle.query(statusQuery);
console.log(`Current status: ${status}`);

// 취소 시그널 전송
await handle.signal(cancelSignal, { reason: 'User requested cancellation' });

// 완료 대기
const result = await handle.result();
console.log(`Final result: ${result}`);

Cron 워크플로우와 Heartbeat 패턴
// Cron 워크플로우 시작 (매일 오전 9시)
await client.workflow.start(dailyReportWorkflow, {
  taskQueue: 'reports',
  workflowId: 'daily-report',
  cronSchedule: '0 9 * * *', // Cron 표현식
});

// 장기 실행 Activity에서 Heartbeat 전송
import { Context } from '@temporalio/activity';

export async function processLargeDataset(datasetId: string): Promise<void> {
  const items = await fetchDataset(datasetId);

  for (let i = 0; i < items.length; i++) {
    // 취소 시그널 체크 (Heartbeat 내부에서 CancelledFailure 발생)
    Context.current().heartbeat(`Processing ${i}/${items.length}`);

    await processItem(items[i]);

    // 워커 재시작 시 마지막 Heartbeat details로 이어서 처리 가능
    const lastHeartbeat = Context.current().info.heartbeatDetails;
    if (lastHeartbeat && i < lastHeartbeat.index) continue;
  }
}