アーキテクチャ

graph TD
    USER["ユーザー"]:::user --> APIGW(["API Gateway"]):::apigw
    APIGW --> PRESIGN["Lambda<br/>(Presigned URL生成)"]:::lambda
    PRESIGN --> S3[("S3<br/>(raw-images/)")]:::storage
    S3 -->|S3 Event| ORCH["Lambda<br/>(Orchestrator)"]:::lambda
    ORCH --> SF(["Step Functions"]):::step

    SF --> REK["1. Rekognition<br/>(コンテンツ検査)"]:::ai
    REK -->|安全| PARALLEL["並列実行"]:::parallel
    REK -->|不適切| BLOCK["SNS通知 + S3削除"]:::error

    PARALLEL --> R1["2. Lambda<br/>(リサイズ 150x150)"]:::lambda
    PARALLEL --> R2["3. Lambda<br/>(リサイズ 600x600)"]:::lambda
    PARALLEL --> R3["4. Lambda<br/>(リサイズ 1200x1200)"]:::lambda

    R1 --> FINISH["5. Lambda<br/>(DB更新 + SES通知)"]:::lambda
    R2 --> FINISH
    R3 --> FINISH

    classDef user fill:#ECF0F1,stroke:#7F8C8D,color:#2C3E50
    classDef apigw fill:#E74C3C,stroke:#A93226,color:#FFFFFF
    classDef lambda fill:#F39C12,stroke:#B7770D,color:#FFFFFF
    classDef storage fill:#3498DB,stroke:#21618C,color:#FFFFFF
    classDef step fill:#9B59B6,stroke:#6C3483,color:#FFFFFF
    classDef ai fill:#1ABC9C,stroke:#148F77,color:#FFFFFF
    classDef parallel fill:#2ECC71,stroke:#1E8449,color:#FFFFFF
    classDef error fill:#E74C3C,stroke:#A93226,color:#FFFFFF

サービス選定理由

エラーハンドリング

// Step Functions の定義（ASL）
const definition = {
  StartAt: "ContentModeration",
  States: {
    ContentModeration: {
      Type: "Task",
      Resource: "arn:aws:lambda:...:content-moderation",
      Retry: [
        {
          ErrorEquals: ["ServiceUnavailable"],
          IntervalSeconds: 2,
          MaxAttempts: 3,
          BackoffRate: 2.0,
        },
      ],
      Catch: [
        {
          ErrorEquals: ["UnsafeContentError"],
          Next: "HandleUnsafeContent",
        },
        {
          ErrorEquals: ["States.ALL"],
          Next: "HandleError",
        },
      ],
      Next: "ParallelResize",
    },
    ParallelResize: {
      Type: "Parallel",
      Branches: [
        { StartAt: "ResizeSmall", States: { /* ... */ } },
        { StartAt: "ResizeMedium", States: { /* ... */ } },
        { StartAt: "ResizeLarge", States: { /* ... */ } },
      ],
      Next: "UpdateAndNotify",
    },
    HandleUnsafeContent: {
      Type: "Task",
      Resource: "arn:aws:lambda:...:delete-and-notify",
      End: true,
    },
    HandleError: {
      Type: "Task",
      Resource: "arn:aws:lambda:...:error-handler",
      End: true,
    },
    UpdateAndNotify: {
      Type: "Task",
      Resource: "arn:aws:lambda:...:update-db-and-notify",
      End: true,
    },
  },
};

コスト見積もり（月間 150,000 枚）

Rekognition が最もコストが大きい。閾値判定を先にLambdaで簡易チェックし、疑わしいもののみ Rekognition に送ることでコスト削減が可能。

アーキテクチャ

graph TD
    API["投稿API"]:::api --> APIGW(["API Gateway"]):::apigw
    APIGW --> SAVE["Lambda<br/>(投稿保存)"]:::lambda
    SAVE --> EB{{"EventBridge"}}:::eventbridge
    EB --> FANOUT["Lambda<br/>(Fan-out Controller)"]:::lambda
    FANOUT --> SQS(["SQS<br/>(バッチキュー)<br/>バッチサイズ: 500件"]):::queue
    FANOUT --> DYNAMO[("DynamoDB<br/>(配信ステータス追跡)")]:::db
    SQS --> WORKER["Lambda<br/>(通知送信Worker)"]:::lambda
    WORKER --> PUSH["SNS Platform /<br/>WebSocket"]:::notify

    classDef api fill:#ECF0F1,stroke:#7F8C8D,color:#2C3E50
    classDef apigw fill:#E74C3C,stroke:#A93226,color:#FFFFFF
    classDef lambda fill:#F39C12,stroke:#B7770D,color:#FFFFFF
    classDef eventbridge fill:#8E44AD,stroke:#6C3483,color:#FFFFFF
    classDef queue fill:#E67E22,stroke:#D35400,color:#FFFFFF
    classDef db fill:#1ABC9C,stroke:#148F77,color:#FFFFFF
    classDef notify fill:#2ECC71,stroke:#1E8449,color:#FFFFFF

Fan-out 戦略

// Fan-out Controller: フォロワーを分割してSQSに投入
export const fanOutHandler = async (event: EventBridgeEvent<"PostCreated", PostDetail>) => {
  const { postId, userId } = event.detail;

  // フォロワーをページネーションで取得
  let lastKey: string | undefined;
  do {
    const followers = await getFollowers(userId, lastKey, 500);

    // 500件ずつSQSにバッチ送信
    await sqs.send(new SendMessageBatchCommand({
      QueueUrl: process.env.NOTIFICATION_QUEUE_URL,
      Entries: [{
        Id: `batch-${lastKey ?? "first"}`,
        MessageBody: JSON.stringify({
          postId,
          followerIds: followers.items.map(f => f.followerId),
        }),
        MessageGroupId: postId,  // FIFO キューで重複防止
        MessageDeduplicationId: `${postId}-${lastKey ?? "first"}`,
      }],
    }));

    lastKey = followers.lastKey;
  } while (lastKey);
};

// 通知送信 Worker
export const notificationWorker = async (event: SQSEvent) => {
  for (const record of event.Records) {
    const { postId, followerIds } = JSON.parse(record.body);

    // バッチで通知送信
    const results = await Promise.allSettled(
      followerIds.map((id: string) => sendPushNotification(id, postId))
    );

    // 配信ステータスを記録
    const succeeded = results.filter(r => r.status === "fulfilled").length;
    const failed = results.filter(r => r.status === "rejected").length;

    await dynamodb.send(new UpdateCommand({
      TableName: process.env.DELIVERY_TABLE,
      Key: { postId, batchId: record.messageId },
      UpdateExpression: "SET succeeded = :s, failed = :f, processedAt = :t",
      ExpressionAttributeValues: {
        ":s": succeeded,
        ":f": failed,
        ":t": new Date().toISOString(),
      },
    }));
  }
};

重複防止

Step Functions による Orchestration Saga

// Step Functions 定義（CDK）
import * as sfn from "aws-cdk-lib/aws-stepfunctions";
import * as tasks from "aws-cdk-lib/aws-stepfunctions-tasks";

const reserveInventory = new tasks.LambdaInvoke(this, "ReserveInventory", {
  lambdaFunction: inventoryFn,
  payload: sfn.TaskInput.fromObject({
    action: "RESERVE",
    "orderId.$": "$.orderId",
    "items.$": "$.items",
  }),
  resultPath: "$.inventoryResult",
});

const releaseInventory = new tasks.LambdaInvoke(this, "ReleaseInventory", {
  lambdaFunction: inventoryFn,
  payload: sfn.TaskInput.fromObject({
    action: "RELEASE",
    "orderId.$": "$.orderId",
    "reservationId.$": "$.inventoryResult.Payload.reservationId",
  }),
});

const processPayment = new tasks.LambdaInvoke(this, "ProcessPayment", {
  lambdaFunction: paymentFn,
  resultPath: "$.paymentResult",
});

const refundPayment = new tasks.LambdaInvoke(this, "RefundPayment", {
  lambdaFunction: paymentFn,
  payload: sfn.TaskInput.fromObject({
    action: "REFUND",
    "paymentId.$": "$.paymentResult.Payload.paymentId",
  }),
});

const arrangeShipping = new tasks.LambdaInvoke(this, "ArrangeShipping", {
  lambdaFunction: shippingFn,
  resultPath: "$.shippingResult",
});

const sendNotification = new tasks.LambdaInvoke(this, "SendNotification", {
  lambdaFunction: notificationFn,
});

// 補償トランザクションチェーン
const compensatePaymentAndInventory = refundPayment
  .next(releaseInventory)
  .next(new sfn.Fail(this, "OrderFailed-Shipping", {
    cause: "Shipping arrangement failed",
  }));

const compensateInventory = releaseInventory
  .next(new sfn.Fail(this, "OrderFailed-Payment", {
    cause: "Payment processing failed",
  }));

// メインフロー
const definition = reserveInventory
  .addCatch(new sfn.Fail(this, "OrderFailed-Inventory"), {
    errors: ["States.ALL"],
  })
  .next(
    processPayment.addCatch(compensateInventory, {
      errors: ["States.ALL"],
      resultPath: "$.error",
    })
  )
  .next(
    arrangeShipping.addCatch(compensatePaymentAndInventory, {
      errors: ["States.ALL"],
      resultPath: "$.error",
    })
  )
  .next(sendNotification)
  .next(new sfn.Succeed(this, "OrderCompleted"));

状態追跡

// 各 Lambda で処理状態を DynamoDB に記録
const updateSagaState = async (
  orderId: string,
  step: string,
  status: "PENDING" | "COMPLETED" | "COMPENSATED" | "FAILED"
) => {
  await dynamodb.send(new UpdateCommand({
    TableName: process.env.SAGA_TABLE,
    Key: { orderId },
    UpdateExpression: "SET #step = :status, updatedAt = :now",
    ExpressionAttributeNames: { "#step": step },
    ExpressionAttributeValues: {
      ":status": status,
      ":now": new Date().toISOString(),
    },
  }));
};

ワークロード 1: レポート生成 → コンテナ（ECS Fargate）推奨

# ECS Scheduled Task
Resources:
  ReportTask:
    Type: AWS::ECS::TaskDefinition
    Properties:
      Cpu: "1024"
      Memory: "2048"
      ContainerDefinitions:
        - Name: report-generator
          Image: !Sub "${AWS::AccountId}.dkr.ecr.${AWS::Region}.amazonaws.com/report-gen:latest"
          Environment:
            - Name: BATCH_SIZE
              Value: "10"

  ReportSchedule:
    Type: AWS::Events::Rule
    Properties:
      ScheduleExpression: "cron(0 15 * * ? *)"  # JST 0:00
      Targets:
        - Arn: !GetAtt ECSCluster.Arn
          EcsParameters:
            TaskCount: 10  # 10タスク × 10件ずつ = 100件

ワークロード 2: WebSocket チャット → コンテナ（ECS Fargate）推奨

API Gateway WebSocket API + Lambda でも実装可能だが、接続管理の複雑さとレイテンシ要件を考慮するとコンテナが適切。

ワークロード 3: Webhook 受信 → サーバーレス（Lambda）推奨

// Lambda Webhook Handler
export const handler = async (event: APIGatewayProxyEvent) => {
  // 署名検証
  const signature = event.headers["x-webhook-signature"];
  if (!verifySignature(event.body!, signature!)) {
    return { statusCode: 401, body: "Invalid signature" };
  }

  // 冪等性チェック
  const webhookId = event.headers["x-webhook-id"];
  const exists = await checkIdempotencyKey(webhookId!);
  if (exists) {
    return { statusCode: 200, body: "Already processed" };
  }

  // 処理（SQS に委譲して即座に応答）
  await sqs.send(new SendMessageCommand({
    QueueUrl: process.env.WEBHOOK_QUEUE_URL,
    MessageBody: event.body!,
    MessageDeduplicationId: webhookId,
  }));

  return { statusCode: 202, body: JSON.stringify({ received: true }) };
};

判断サマリ