AWS Lambda Durable Functions を試す : APIによる非同期処理のウェイトとリスタート

import { withDurableExecution } from '@aws/durable-execution-sdk-js';

export const handler = withDurableExecution(async (event, context) => {
  // TODO implement

  await context.step('Step #1', (stepCtx) => {
    stepCtx.logger.info('Hello from step #1');
  });

  // Pause for 1 second without consuming CPU cycles or incurring usage charges
  await context.wait({ seconds:1 });

  // Context logger is replay aware and will not log the same message multiple times
  context.logger.info('Waited for 1 second');

  const message = await context.step('Step #2', async () => {
    return 'Hello from Durable Lambda!';
  });

  const response = {
    statusCode: 200,
    body: JSON.stringify(message),
  };
  return response;
});

  await context.step('Step #1', (stepCtx) => {
    stepCtx.logger.info('Hello from step #1');
  });

  await context.wait({ seconds:1 });

  const message = await context.step('Step #2', async () => {
    return 'Hello from Durable Lambda!';
  }); 

import { withDurableExecution } from "@aws/durable-execution-sdk-js";

export const handler = withDurableExecution(
  async (event, context) => {
    const { username, email } = event;
    
    context.logger.info(`ユーザー登録処理開始: ${username}`);
    
    // Step 1: ユーザー情報を検証
    const validationResult = await context.step(async (stepContext) => {
      stepContext.logger.info(`ユーザー検証中: ${username}`);
      
      if (!username || !email) {
        throw new Error("ユーザー名とメールアドレスは必須です");
      }
      
      return {
        username,
        email,
        status: "validated"
      };
    });
    
    context.logger.info(`検証完了: ${JSON.stringify(validationResult)}`);
    
    // Step 2: 外部APIに処理リクエストを送信（非同期処理を開始）
    const jobId = await context.step(async (stepContext) => {
      const url = "https://kameda.requestcatcher.com/start-job";
      
      const data = {
        username,
        email,
        timestamp: new Date().toISOString(),
        message: "非同期処理を開始"
      };
      
      try {
        const response = await fetch(url, {
          method: 'POST',
          headers: {
            'Content-Type': 'application/json'
          },
          body: JSON.stringify(data)
        });
        
        stepContext.logger.info(`処理開始リクエスト送信: ${response.status}`);
        
        // 実際のAPIではジョブIDが返されることを想定
        // ここでは仮のIDを生成
        const jobId = `job-${Date.now()}`;
        
        return {
          jobId: jobId,
          startTime: new Date().toISOString()
        };
      } catch (error) {
        stepContext.logger.error(`APIエラー: ${error.message}`);
        throw error;
      }
    });
    
    context.logger.info(`ジョブID: ${jobId.jobId} - ポーリング開始`);
    
    // Step 3: ポーリングで結果を確認（最大10回、各5秒待機）
    const maxAttempts = 10;
    const pollIntervalSeconds = 5;
    let jobCompleted = false;
    let jobResult = null;
    
    for (let attempt = 1; attempt <= maxAttempts; attempt++) {
      context.logger.info(`ポーリング試行 ${attempt}/${maxAttempts}`);
      
      // 5秒待機
      await context.wait({ seconds: pollIntervalSeconds });
      
      // 外部APIのステータスを確認
      const statusCheck = await context.step(async (stepContext) => {
        const statusUrl = `https://kameda.requestcatcher.com/check-status/${jobId.jobId}`;
        
        try {
          const response = await fetch(statusUrl, {
            method: 'GET',
            headers: {
              'Content-Type': 'application/json'
            }
          });
          
          stepContext.logger.info(`ステータス確認: ${response.status}`);
          
          // 実際のAPIでは、レスポンスボディにステータスが含まれる
          // ここでは仮の判定：200なら完了と見なす
          if (response.status === 200) {
            return {
              completed: true,
              status: "success",
              data: {
                username,
                email,
                processedAt: new Date().toISOString()
              }
            };
          } else {
            return {
              completed: false,
              status: "processing",
              attempt: attempt
            };
          }
        } catch (error) {
          stepContext.logger.error(`ステータス確認エラー: ${error.message}`);
          return {
            completed: false,
            status: "error",
            error: error.message
          };
        }
      });
      
      context.logger.info(`ステータス確認結果: ${JSON.stringify(statusCheck)}`);
      
      if (statusCheck.completed) {
        jobCompleted = true;
        jobResult = statusCheck;
        context.logger.info(`処理完了を確認しました（試行${attempt}回目）`);
        break;
      }
      
      context.logger.info(`まだ処理中... 次の確認まで${pollIntervalSeconds}秒待機`);
    }
    
    // Step 4: タイムアウトチェック
    if (!jobCompleted) {
      throw new Error(`タイムアウト: ${maxAttempts * pollIntervalSeconds}秒以内に処理が完了しませんでした`);
    }
    
    // Step 5: ウェルカムメッセージを送信
    const welcomeResult = await context.step(async (stepContext) => {
      stepContext.logger.info(`ウェルカムメッセージ送信: ${email}`);
      
      const message = `ようこそ、${username}さん！外部システムの処理が完了し、登録が完了しました。`;
      
      return {
        username,
        email,
        message,
        status: "sent"
      };
    });
    
    context.logger.info(`メッセージ送信完了: ${JSON.stringify(welcomeResult)}`);
    
    // 最終結果を返す
    return {
      status: "completed",
      username,
      email,
      message: welcomeResult.message,
      jobResult: jobResult,
      totalWaitTime: `約${pollIntervalSeconds * (jobResult.attempt || 1)}秒`
    };
  }
);