第27章: R2 Event Notifications によるイベント連鎖（実践チュートリアル）

AWSの「S3 Event Notifications + SQS + Lambda」に相当するイベント駆動パイプラインが、Cloudflareでは R2（オブジェクトストレージ）＋ Queues（メッセージキュー）＋ Workers の3点セットで、超低コスト・超低遅延に構築できます。

本章では、「ユーザーがR2に画像ファイルをアップロードした瞬間、自動でイベントがキューに届き、後続のWorkerがサムネイル生成や画像解析を行う」 自動処理パイプラインをハンズオンで作成します。

本チュートリアルのゴール

R2バケットとキューの作成（CLI）
wrangler r2 bucket notification create によるオブジェクト作成イベントの紐付け
イベントペイロードを非同期受信して処理するWorkerの実装

【イベント連鎖の全体像】
[ ユーザー / アプリ ] ──ファイルアップロード──> [ R2 バケット (user-media) ]
                                                        │
                                                        ▼ オブジェクト作成イベント (object-create)
                                                [ Cloudflare Queues (r2-upload-events) ]
                                                        │
                                                        ▼ 自動エンキュー
                                                [ イベント処理Worker ] (サムネイル生成・通知)

Step 1: バケットとキューの作成（CLI）

# 1. アップロード用R2バケットの作成
npx wrangler r2 bucket create user-media

# 2. イベント通知を受け取るキューの作成
npx wrangler queues create r2-upload-events

Step 2: R2イベント通知の登録（CLI）

特定バケットに対して、ファイル作成イベント（object-create）をキューへ転送するルールを登録します。

npx wrangler r2 bucket notification create user-media \
  --event-types object-create \
  --queue r2-upload-events \
  --prefix "uploads/"

確認（登録済みルールの一覧表示）:

npx wrangler r2 bucket notification list user-media

これで、uploads/ ディレクトリ配下にファイルが置かれると、自動的にキューへイベントが発行されます。

Step 3: `wrangler.jsonc` の設定

Workerがキューを購読（Consume）し、かつR2バケットから画像を取り出せるよう設定します。

{
  "name": "r2-event-consumer",
  "main": "src/index.ts",
  "compatibility_date": "2024-09-01",
  "compatibility_flags": [
    "nodejs_compat"
  ],
  "r2_buckets": [
    {
      "binding": "MEDIA_BUCKET",
      "bucket_name": "user-media"
    }
  ],
  "queues": {
    "consumers": [
      {
        "queue": "r2-upload-events",
        "max_batch_size": 10,
        "max_batch_timeout": 5
      }
    ]
  }
}

型定義を更新します。

npx wrangler types

Step 4: イベント処理Workerの実装（`src/index.ts`）

// R2イベント通知ペイロードの型定義
type R2EventPayload = {
  account: string;
  action: string;
  bucket: string;
  object: {
    key: string;
    size: number;
    eTag: string;
  };
  eventTime: string;
};

export default {
  // キューからバッチでメッセージを取り出すハンドラー
  async queue(batch: MessageBatch<R2EventPayload>, env: Env): Promise<void> {
    console.log(`[イベント受信] ${batch.messages.length} 件のファイルイベントを処理します`);

    for (const message of batch.messages) {
      const event = message.body;
      const objectKey = event.object.key;
      const bucketName = event.bucket;
      const fileSizeKb = Math.round(event.object.size / 1024);

      console.log(`新規ファイル検知: バケット=${bucketName}, ファイル名=${objectKey}, サイズ=${fileSizeKb} KB`);

      try {
        // R2から実際にファイルの中身を取得
        const fileObj = await env.MEDIA_BUCKET.get(objectKey);
        if (fileObj) {
          // ここでサムネイル生成、画像圧縮、AI OCR分析などを実行
          console.log(`[処理成功] ${objectKey} の解析・サムネイル生成が完了しました`);
        }

        // メッセージの処理成功を通知（キューから削除）
        message.ack();
      } catch (err) {
        console.error(`[処理失敗] ${objectKey}:`, err);
        // エラー時は再試行（リトライ）
        message.retry();
      }
    }
  },

  // HTTPエンドポイント（手動ヘルスチェック用）
  async fetch(): Promise<Response> {
    return new Response('R2 Event Consumer Worker is active');
  }
};

Step 5: デプロイと動作テスト

# Workerの本番デプロイ
npx wrangler deploy

# ログをリアルタイムストリーミング監視
npx wrangler tail

別のターミナルから、R2バケットにファイルをアップロードしてみます。

# テスト用画像ファイルをアップロード
npx wrangler r2 object put user-media/uploads/photo.jpg --file=./sample.jpg

wrangler tail の出力画面:

[イベント受信] 1 件のファイルイベントを処理します
新規ファイル検知: バケット=user-media, ファイル名=uploads/photo.jpg, サイズ=245 KB
[処理成功] uploads/photo.jpg の解析・サムネイル生成が完了しました

数秒以内にイベントが自動検知され、バックグラウンド処理が完了します。

まとめ

ポーリング不要: 「新着ファイルがあるか」を定期的にチェックする無駄なCronループを廃止。
堅牢な再試行: 処理失敗時は message.retry() で自動的に再試行されるため、データ欠落を防げます。

💡 用語解説コラム

[!NOTE] イベント駆動アーキテクチャ (EDA)
「ファイルが保存された」「データが更新された」という出来事（イベント）をトリガーとして、後続のプログラムを連鎖的に自動実行する疎結合な設計思想。

[!NOTE] ポーリング (Polling)
変化があったかどうかを定期的にサーバーへ問い合わせ続ける方式。イベント駆動に切り替えることで無駄な通信とAPIコストをゼロに削減できます。