Temporal

エンジニアのためのWebチートシート

基本的なWorkflow

• proxyActivitiesでActivityを呼び出すWorkflowの基本パターンです。

  import { proxyActivities, sleep } from '@temporalio/workflow';
  import type * as activities from './activities';
  
  const { processPayment, sendEmail, shipOrder } =
    proxyActivities<typeof activities>({
      startToCloseTimeout: '30s',
      retry: { maximumAttempts: 3 },
    });

  export async function orderWorkflow(order: OrderInput) {
    await processPayment(order);
    await sendEmail(order.email, 'Order confirmed!');
    await sleep('2 hours');  // Temporal timer (not setTimeout)
    return await shipOrder(order);
  }

並列Activity実行

• Promise.allで複数のActivityを並列に実行するパターンです。

  export async function checkoutWorkflow(order: OrderInput) {
    // Run validations in parallel
    const [inventory, fraud, address] = await Promise.all([
      validateInventory(order.items),
      checkFraud(order.payment),
      verifyAddress(order.address),
    ]);
  
    if (!inventory.available) throw new Error('Out of stock');
    return { status: 'approved', address: address.normalized };
  }

基本的なActivity

• 外部API呼び出しやDB操作を行うActivityの定義です。

  // activities.ts — I/O, API calls, DB operations
  export async function processPayment(order: OrderInput) {
    const res = await fetch('https://api.stripe.com/v1/charges', {
      method: 'POST',
      headers: { Authorization: `Bearer ${process.env.STRIPE_KEY}` },
      body: JSON.stringify({ amount: order.total, currency: 'jpy' }),
    });
    return res.json();
  }
  
  export async function sendEmail(to: string, subject: string) {
    await emailService.send({ to, subject });
  }
  
  export async function shipOrder(order: OrderInput) {
    return (await shippingApi.create(order)).id;
  }

ハートビート

• 長時間実行するActivityで進捗を報告するパターンです。

  import { heartbeat } from '@temporalio/activity';
  
  export async function processLargeFile(filePath: string) {
    const lines = await readLines(filePath);
    for (let i = 0; i < lines.length; i++) {
      await processLine(lines[i]);
      if (i % 100 === 0) heartbeat({ processed: i });
    }
  }

  // Workflow side: set heartbeat timeout
  const { processLargeFile } = proxyActivities({
    startToCloseTimeout: '1 hour',
    heartbeatTimeout: '30s',
  });

Worker起動

• WorkflowとActivityを実行するWorkerの設定です。

  // worker.ts
  import { Worker } from '@temporalio/worker';
  import * as activities from './activities';
  
  const worker = await Worker.create({
    workflowsPath: require.resolve('./workflows'),
    activities,
    taskQueue: 'order-processing',
  });
  await worker.run();

Workflowの開始（Client）

• Temporal Clientを使ってWorkflowを開始するコードです。

  // client.ts
  import { Client } from '@temporalio/client';
  const client = new Client();
  
  // Start a workflow
  const handle = await client.workflow.start(orderWorkflow, {
    taskQueue: 'order-processing',
    workflowId: `order-${orderId}`,
    args: [{ id: orderId, items, email }],
  });
  
  const result = await handle.result();  // Wait for completion

  // Get handle to existing workflow
  const existing = client.workflow.getHandle('order-12345');
  const status = await existing.describe();

シグナル定義

• 外部からWorkflowに送信するシグナルの定義と受信パターンです。

  import { defineSignal, setHandler, condition }
    from '@temporalio/workflow';
  
  export const approvalSignal = defineSignal<[boolean]>('approval');

  export async function approvalWorkflow(req: Request) {
    let approved: boolean | undefined;
  
    setHandler(approvalSignal, (v) => { approved = v; });
  
    // Wait up to 7 days for signal
    await condition(() => approved !== undefined, '7 days');
    return approved ? 'approved' : 'rejected';
  }

クエリ定義

• Workflowの状態を読み取るクエリの定義パターンです。

  import { defineQuery, setHandler } from '@temporalio/workflow';
  
  export const statusQuery = defineQuery<string>('status');
  export const progressQuery = defineQuery<number>('progress');

  export async function processingWorkflow(items: Item[]) {
    let status = 'running', progress = 0;
    setHandler(statusQuery, () => status);
    setHandler(progressQuery, () => progress);
  
    for (const item of items) {
      await processItem(item);
      progress++;
    }
    status = 'completed';
  }

外部からのシグナル送信

• Clientを使ってWorkflowにシグナルを送信するコードです。

  // client-usage.ts
  const handle = client.workflow.getHandle(
    'approval-workflow-123'
  );
  
  // Send signal
  await handle.signal(approvalSignal, true);
  
  // Send query
  const status = await handle.query(statusQuery);
  const progress = await handle.query(progressQuery);

リトライポリシー

• Activity Retry Policy

  const { riskyActivity } = proxyActivities({
    startToCloseTimeout: '30s',
    retry: {
      initialInterval: '1s',
      backoffCoefficient: 2,
      maximumInterval: '30s',
      maximumAttempts: 5,         // 0 = unlimited
      nonRetryableErrorTypes: ['InvalidInputError'],
    },
  });

• Workflow-level Timeout

  const handle = await client.workflow.start(myWorkflow, {
    taskQueue: 'my-queue',
    workflowId: 'wf-1',
    workflowExecutionTimeout: '24 hours',
    workflowRunTimeout: '1 hour',
  });