サーバーレスでステートフルなワークフローを構築する：Lambda Durable Functionsの設計と実装

mkdir lambda-durable-demo
cd lambda-durable-demo
npx cdk init app --language typescript
npm install @aws-cdk/aws-lambda-durable

const durableFunctionRole = new iam.Role(this, 'DurableFunctionRole', {
  assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
  managedPolicies: [
    iam.ManagedPolicy.fromAwsManagedPolicyName('service-role/AWSLambdaBasicExecutionRole'),
  ],
});

// DynamoDBへのアクセス権限を追加
durableFunctionRole.addToPolicy(new iam.PolicyStatement({
  actions: ['dynamodb:PutItem', 'dynamodb:GetItem', 'dynamodb:UpdateItem', 'dynamodb:Query'],
  resources: ['arn:aws:dynamodb:*:*:table/DurableTaskState*'],
}));

import { DurableOrchestrationContext } from '@aws-lambda-durable/core';

export async function orchestrator(context: DurableOrchestrationContext): Promise {
  // Step 1: データ取得
  const rawData = await context.callActivity('fetchData', { source: 'api' });
  
  // Step 2: データ変換
  const transformedData = await context.callActivity('transformData', { data: rawData });
  
  // Step 3: 結果保存
  const result = await context.callActivity('saveResult', { data: transformedData });
  
  return `Workflow completed: ${result.recordCount} records processed`;
}

// fetchData Activity
export async function fetchData(input: { source: string }): Promise {
  const response = await fetch(`https://api.example.com/data?source=${input.source}`);
  return response.json();
}

// transformData Activity
export async function transformData(input: { data: RawData[] }): Promise {
  return input.data.map(item => ({
    id: item.id,
    value: item.value * 1.1,
    processedAt: new Date().toISOString(),
  }));
}

// saveResult Activity
export async function saveResult(input: { data: TransformedData[] }): Promise<{ recordCount: number }> {
  // DynamoDBに保存する処理
  await dynamoDb.batchWrite({ /* ... */ });
  return { recordCount: input.data.length };
}

export async function parallelOrchestrator(context: DurableOrchestrationContext): Promise {
  // 3つのAPIを並列に呼び出し
  const tasks = [
    context.callActivity('fetchFromApiA', {}),
    context.callActivity('fetchFromApiB', {}),
    context.callActivity('fetchFromApiC', {}),
  ];
  
  // すべての完了を待機
  const results = await context.all(tasks);
  
  // 結果を集約
  const aggregated = await context.callActivity('aggregateResults', { results });
  return aggregated;
}

export async function timerOrchestrator(context: DurableOrchestrationContext): Promise {
  // 外部システムにリクエスト送信
  const requestId = await context.callActivity('sendExternalRequest', {});
  
  // 5分待機
  await context.createTimer(new Date(Date.now() + 5 * 60 * 1000));
  
  // 結果を取得
  const result = await context.callActivity('fetchExternalResult', { requestId });
  return result;
}

export async function retryOrchestrator(context: DurableOrchestrationContext): Promise {
  const retryOptions = {
    maxRetries: 3,
    initialDelay: 1000,
    backoffCoefficient: 2,
  };
  
  try {
    const result = await context.callActivityWithRetry('unreliableActivity', {}, retryOptions);
    return result;
  } catch (error) {
    // 3回リトライしても失敗した場合のフォールバック
    await context.callActivity('notifyFailure', { error: error.message });
    throw error;
  }
}