Temporal로 장기 실행 비즈니스 워크플로 구현하기

pnpm add @temporalio/client @temporalio/worker @temporalio/workflow @temporalio/activity

// greeting.activities.ts
export async function greet(name: string): Promise<string> {
  return `안녕하세요, ${name}!`;
}

// greeting.workflow.ts
import { proxyActivities } from '@temporalio/workflow';
import type * as activities from './greeting.activities';
 
const { greet } = proxyActivities<typeof activities>({
  startToCloseTimeout: '5s',
});
 
export async function greetingWorkflow(name: string): Promise<string> {
  return await greet(name);
}

// client.ts
import { Client, Connection } from '@temporalio/client';
import { greetingWorkflow } from './greeting.workflow';
 
const connection = await Connection.connect();
const client = new Client({ connection });
 
const result = await client.workflow.execute(greetingWorkflow, {
  args: ['세상'],
  taskQueue: 'greeting',
  workflowId: 'greeting-001',
});
console.log(result); // "안녕하세요, 세상!"

// payment.workflow.ts
import { proxyActivities, ApplicationFailure } from '@temporalio/workflow';
import type * as activities from './payment.activities';
 
const {
  reserveInventory,
  chargePayment,
  sendConfirmation,
  releaseInventory,
  refundPayment,
} = proxyActivities<typeof activities>({
  startToCloseTimeout: '30s',
  retry: {
    maximumAttempts: 5,
    initialInterval: '1s',
    backoffCoefficient: 2,
    nonRetryableErrorTypes: ['InvalidPaymentMethodError'],
  },
});
 
export async function orderWorkflow(orderId: string, amount: number): Promise<void> {
  const compensations: Array<() => Promise<void>> = [];
 
  try {
    await reserveInventory(orderId);
    compensations.push(() => releaseInventory(orderId));
 
    await chargePayment(orderId, amount);
    compensations.push(() => refundPayment(orderId));
 
    await sendConfirmation(orderId);
    // 메일 자체는 되돌릴 수 없어 보상을 쌓지 않는다.
    // 이 단계에서 예외가 발생하면 catch가 앞선 결제·재고 보상을 정상 실행한다.
 
  } catch (originalErr) {
    // 각 보상을 독립적으로 실행 — 하나가 실패해도 나머지는 계속
    const compensationErrors: Error[] = [];
    for (const compensate of [...compensations].reverse()) {
      try {
        await compensate();
      } catch (err) {
        compensationErrors.push(err as Error);
      }
    }
 
    if (compensationErrors.length > 0) {
      // 부분 롤백 상태: 워크플로는 FAILED로 종료되고 수동 개입이 필요하다
      throw ApplicationFailure.nonRetryable(
        `Saga 보상 ${compensationErrors.length}건 최종 실패 — 수동 복구 필요`,
        'PartialCompensationFailure'
      );
    }
 
    throw originalErr;
  }
}

// payment.activities.ts
import { stripe } from './stripe.client';
import { db } from './db.client';
 
export async function chargePayment(orderId: string, amount: number): Promise<string> {
  const charge = await stripe.charges.create(
    { amount, currency: 'usd', source: 'tok_visa' },
    { idempotencyKey: `charge-${orderId}` }
  );
 
  // 결제 기록 저장 — refundPayment에서 stripeId를 조회한다
  await db.charge.upsert({
    where: { orderId },
    create: { orderId, stripeId: charge.id },
    update: {}, // 이미 저장됐으면 무시 (멱등)
  });
 
  return charge.id;
}
 
export async function reserveInventory(orderId: string): Promise<void> {
  await db.inventory.upsert({
    where: { orderId },
    create: { orderId, status: 'reserved' },
    update: {},
  });
}
 
export async function releaseInventory(orderId: string): Promise<void> {
  await db.inventory.updateMany({
    where: { orderId, status: 'reserved' },
    data: { status: 'released' },
  });
}
 
export async function refundPayment(orderId: string): Promise<void> {
  const charge = await db.charge.findUnique({ where: { orderId } });
  if (!charge) return; // chargePayment가 DB 저장 전 실패한 경우
 
  await stripe.refunds.create(
    { charge: charge.stripeId },
    { idempotencyKey: `refund-${orderId}` }
  );
}
 
export async function sendConfirmation(orderId: string): Promise<void> {
  await emailService.send({
    template: 'order-confirmed',
    orderId,
    idempotencyKey: `email-${orderId}`,
  });
}

// shipping.workflow.ts
import {
  proxyActivities,
  sleep,
  now,
  workflowInfo,
  continueAsNew,
  ApplicationFailure,
} from '@temporalio/workflow';
import type * as activities from './shipping.activities';
 
const { checkShipmentStatus, notifyDelivered } =
  proxyActivities<typeof activities>({ startToCloseTimeout: '10s' });
 
export interface TrackShipmentInput {
  orderId: string;
  deadlineEpoch: number; // 클라이언트에서 Date.now() + N 일로 계산해 전달
}
 
export async function trackShipmentWorkflow({
  orderId,
  deadlineEpoch,
}: TrackShipmentInput): Promise<void> {
  while (true) {
    if (now() >= deadlineEpoch) {
      throw ApplicationFailure.nonRetryable(`배송 타임아웃: ${orderId}`);
    }
 
    const status = await checkShipmentStatus(orderId);
 
    if (status === 'delivered') {
      await notifyDelivered(orderId);
      return;
    }
 
    if (status === 'lost' || status === 'returned') {
      throw ApplicationFailure.nonRetryable(`배송 이상: ${orderId} - ${status}`);
    }
 
    // 히스토리가 커지기 전에 새 실행으로 이어받는다
    // deadlineEpoch(절대값)를 넘기므로 재시작 후에도 원래 데드라인이 유지된다
    if (workflowInfo().historyLength > 1000) {
      await continueAsNew<typeof trackShipmentWorkflow>({ orderId, deadlineEpoch });
    }
 
    await sleep('30m');
  }
}

await client.workflow.start(trackShipmentWorkflow, {
  args: [{ orderId, deadlineEpoch: Date.now() + 14 * 24 * 60 * 60 * 1000 }],
  taskQueue: 'shipping',
  workflowId: `shipping-${orderId}`,
});

// approval.workflow.ts
import {
  defineSignal,
  setHandler,
  condition,
  proxyActivities,
} from '@temporalio/workflow';
import type * as activities from './approval.activities';
 
export const approvalSignal = defineSignal<[boolean]>('approvalDecision');
 
const { processApprovedOrder, notifyRejection } =
  proxyActivities<typeof activities>({ startToCloseTimeout: '30s' });
 
export async function manualApprovalWorkflow(orderId: string): Promise<string> {
  let approved: boolean | null = null;
 
  setHandler(approvalSignal, (decision: boolean) => {
    approved = decision;
  });
 
  // 최대 48시간 대기, 응답 없으면 false 반환
  const received = await condition(() => approved !== null, '48h');
 
  if (!received || approved === false) {
    await notifyRejection(orderId);
    return 'rejected';
  }
 
  await processApprovedOrder(orderId);
  return 'approved';
}

// 클라이언트 코드
import { Client } from '@temporalio/client';
import { approvalSignal } from './approval.workflow';
 
const client = new Client();
await client.workflow.getHandle(`approval-${orderId}`).signal(approvalSignal, true);

// worker.ts
import { Worker } from '@temporalio/worker';
import * as paymentActivities from './payment.activities';
import * as shippingActivities from './shipping.activities';
 
const worker = await Worker.create({
  workflowsPath: new URL('./workflows', import.meta.url).pathname, // ESM 환경
  activities: {
    ...paymentActivities,
    ...shippingActivities,
  },
  taskQueue: 'order-processing',
});
 
await worker.run();

// 잘못된 예
export async function wrongWorkflow(orderId: string) {
  const response = await fetch(`https://api.example.com/order/${orderId}`);
}
 
// 올바른 예
export async function correctWorkflow(orderId: string) {
  const result = await fetchOrderDetails(orderId); // proxyActivities로 감싼 Activity
}

// 잘못된 예 — Worker 재시작 시 타이머 소멸
await new Promise(resolve => setTimeout(resolve, 30000));
 
// 올바른 예 — Worker가 죽어도 타이머 상태 유지
await sleep('30s');