A+B 複合 — コマンド・クエリ分離 × フラグ引数の排除 × null非返却（商品画像アップロード後処理 Bad→Good Ch13）× Cloud Run + Eventarc 画像処理ワーカーの冪等性(event_id dedupe) × 最小権限(roles/editor廃止) × リソース適正化(OOMKilled解消) × Pub/Subデッドレターキュー × ingress制限

2026-07-25 (Day 108) 土曜複合問題 ★★★★☆ Python 3.12 / frozen dataclass / Protocol / CQS Cloud Run v2 / Eventarc / Pub/Sub DLQ / Terraform

概要

🚩

フラグ引数は「内部にif文を隠す」アンチパターン

resize/watermark/thumbnail/notify の4つの真偽値は 2^4=16通り の内部分岐を1関数に押し込めていた。ImageProcessor にメソッドを分割し、呼び出し側がビジネスルールに基づいて必要なメソッドだけを呼ぶ設計に置き換える。

🚫

nullは「失敗」と「意図的に何もしない」を区別できない

元のコードは両ケースで None を返し、呼び出し側は一律 error 扱いにするしかなかった。専用例外 ImageLoadError に統一し、フェイルファストにする（Ch13: null返却の禁止）。

🔀

コマンド・クエリ分離（CQS）で戻り値の意図を明確にする

保存・通知という副作用（コマンド）を実行する関数が、無関係な処理件数（クエリ的な値）を返していた。save() は保存先パスのみを返すコマンドとして再設計する。

♻️

at-least-once配信を前提に冪等性を組み込む

Eventarc（裏側はPub/Sub）の配信保証は「最低1回」。event_id ベースのdedupeチェックなしでは重複処理・重複通知が必ず発生する。冪等性はイベント駆動アーキテクチャの必須要件。

問題 A: コーディング — コマンド・クエリ分離 × フラグ引数の排除 × null非返却（商品画像アップロード後処理 Bad→Good）

以下の「悪いコード」は、ECサイト MOps チームが運用する商品画像アップロード後処理（リサイズ・透かし・サムネイル生成・Slack通知）の関数です。問題点を全て洗い出し、「良いコード・悪いコードで学ぶ設計入門」第13章（メソッド設計: コマンド・クエリ分離／フラグ引数の排除／null返却の禁止）を使って Bad→Good にリファクタリングしてください。

制約・前提条件

  • Python 3.12+
  • resize/watermark/thumbnail/notify のようなフラグ引数を排除し、呼び出し側が必要なメソッドだけを明示的に呼び出す設計にすること
  • 「何もしない」場合と「失敗した」場合の両方を None で表現しないこと。失敗は専用例外を送出すること
  • 画像の加工（リサイズ・透かし・保存・サムネイル生成）はコマンド（副作用あり・戻り値は最小限）とし、無関係な値（処理件数など）を返さないこと（コマンド・クエリ分離、CQS）
  • 通知の要否判断は関数内部のフラグではなく、呼び出し側の責務とすること
  • モジュールグローバル変数による状態共有を廃止し、状態はインスタンスに閉じ込めること（完全コンストラクタ）
  • Google スタイル docstring・インラインコメント・名前付き定数を含めること
期待する回答形式: 問題点の列挙（番号付き）+ 改善後コード + 実行例（input→output）+ 適用した設計パターン名と書籍対応章

悪いコード (Before) — カテゴリ A

このコードには 7つの設計上の問題 が隠れています。見つけてみてください。
bad_image_processor.py — グローバル状態・フラグ引数爆発・null二重使い・CQS違反
processed_count = 0  # 問題1: モジュールグローバル変数に処理件数を保持(隠れた状態)


def process_image(image_path, resize=True, watermark=False, thumbnail=False, notify=True):
    global processed_count

    if not resize and not watermark and not thumbnail:
        return None  # 問題3: 「何もしない」ケースをNoneで表現(失敗と区別できない)

    try:
        img = _load_image(image_path)
    except Exception as e:
        print(f"failed to load: {e}")  # 問題4: 例外を握りつぶし、ログレベルも不明
        return None  # 問題3(続き): 失敗時も同じNoneを返す

    if resize:
        img = _resize(img, max_width=1200)
    if watermark:
        img = _apply_watermark(img)

    if thumbnail:
        thumb = _make_thumbnail(img, size=200)
        _save(thumb, image_path.replace(".jpg", "_thumb.jpg"))  # 問題6: 拡張子.jpg決め打ち

    _save(img, image_path)
    processed_count += 1  # 問題1(続き): グローバル状態を書き換える副作用

    if notify:
        _send_slack_notification(f"{image_path} processed")  # 問題7: 通知要否もフラグに混在

    return processed_count  # 問題5: 保存・通知(コマンド)なのに件数(クエリ的な値)を返す


# 呼び出し側 — Cloud Run(Eventarcトリガー)のHTTPハンドラ
def handle_upload_event(request):
    data = request.get_json()
    image_path = data["name"]
    category = data.get("category", "default")

    needs_thumbnail = category in ("apparel", "shoes")
    result = process_image(
        image_path,
        resize=True,
        watermark=True,
        thumbnail=needs_thumbnail,  # 問題2: フラグ引数の組み合わせ(2^4=16通り)を1関数に集約
        notify=True,
    )
    if result is None:
        return "error", 500  # 問題3(続き): 呼び出し側もNoneの意味を判別できない
    return "ok", 200
問題点サマリー（7点）
1グローバル変数processed_countによる隠れた状態（Ch13） — 誰がいつ書き換えたか追跡不能。状態はインスタンスに閉じ込めるべき
2フラグ引数の組み合わせ爆発（Ch13） — 4フラグで2^4=16通りの内部分岐を1関数に集約。使われる組み合わせが読み取れない
3「何もしない」と「失敗」の両方をNoneで表現（Ch13） — 呼び出し側は一律error扱いにするしかない
4例外の握りつぶし（Ch10/Ch13） — printのみで専用例外に変換していない
5コマンド・クエリ分離違反（Ch13） — 保存・通知の副作用なのに件数を返す
6拡張子.jpg決め打ちの文字列操作（Ch10） — PNG等の他フォーマットで壊れる
7通知の要否まで同じ関数のフラグに混在（Ch6-7） — 画像加工と通知が同居し単一責任違反

ヒント A（段階的開示）

ヒント1 — 方向性
process_image は「何をするか」を4つの真偽値フラグで制御しており、呼び出し側からは 2^4=16通り の組み合わせのうちどれが実際に使われるか読み取れない。フラグ引数は「関数の内部にif文で分岐を隠す」アンチパターンであり、対処法は関数を分割し、呼び出し側が必要なメソッドだけを明示的に呼ぶこと。また「何もしない」と「失敗」を同じ None で表現すると、呼び出し側は分岐できない。これは第13章の「nullを返さない」原則に反する。保存・通知という副作用（コマンド）を実行する関数が、無関係な処理件数（クエリ的な値）を返しているのもコマンド・クエリ分離（CQS）違反。
ヒント2 — アプローチ
  • ImageProcessor クラスを作り、resize()/apply_watermark()/save()/save_thumbnail() をそれぞれ独立したメソッドにする（フラグではなく呼び出し側が選んで呼ぶ）
  • 「サムネイルが必要か」は真偽値フラグではなく THUMBNAIL_REQUIRED_CATEGORIES のようなビジネスルール（商品カテゴリ）で判定する
  • 画像読み込み失敗は raise ImageLoadError(...) で表現し、None を返さない
  • save()/save_thumbnail() は保存先の Path のみを返し、処理件数のような無関係な値は返さない
  • Notifier を呼び出し側から注入し、通知の要否・内容は呼び出し側（handle_upload_event）の責務にする
ヒント3 — コードの骨格 
class ImageProcessor:
    def __init__(self, image_path: Path) -> None:
        self._image_path = image_path
        self._image = self._load(image_path)  # 失敗時は例外(Noneを返さない)

    def resize(self, max_width: int = 1200) -> None: ...      # コマンド
    def apply_watermark(self) -> None: ...                     # コマンド
    def save(self) -> Path: ...                                # コマンド(保存先のみ返す)
    def save_thumbnail(self, size: int = 200) -> Path: ...     # コマンド

def handle_upload_event(image_path: Path, category: str, notifier) -> ProcessedImage:
    processor = ImageProcessor(image_path)
    processor.resize()
    processor.apply_watermark()
    main_path = processor.save()
    thumbnail_path = processor.save_thumbnail() if category in THUMBNAIL_REQUIRED_CATEGORIES else None
    notifier.send(f"{main_path} processed")
    return ProcessedImage(main_path=main_path, thumbnail_path=thumbnail_path)

問題点分析 — カテゴリ A

#問題点分類改善方法
1グローバル変数processed_countによる隠れた状態完全コンストラクタ Ch13状態をImageProcessorインスタンスに閉じ込める
2フラグ引数(4つ)の組み合わせ爆発フラグ引数の排除 Ch13メソッド分割+呼び出し側が明示的に選択
3「何もしない」と「失敗」の両方でNoneを返却null返却の禁止 Ch13専用例外ImageLoadErrorに統一
4例外の握りつぶし設計の悪魔 Ch10専用例外+loggingに置き換え
5コマンド・クエリ分離違反(副作用が件数を返す)CQS Ch13save()は保存先パスのみ返却
6拡張子.jpg決め打ちの文字列操作設計の悪魔 Ch10Path.with_stemで拡張子非依存に
7通知要否がフラグに混在(単一責任違反)関心の分離 Ch6-7Notifier注入+呼び出し側の責務化

模範解答 A

Before — グローバル状態・フラグ引数爆発・null二重使い
processed_count = 0  # 隠れたグローバル状態

def process_image(image_path, resize=True, watermark=False, thumbnail=False, notify=True):
    global processed_count
    if not resize and not watermark and not thumbnail:
        return None                       # 「何もしない」もNone
    try:
        img = _load_image(image_path)
    except Exception as e:
        print(f"failed to load: {e}")     # 握りつぶし
        return None                       # 「失敗」もNone(区別不能)
    ...
    if thumbnail:
        _save(thumb, image_path.replace(".jpg", "_thumb.jpg"))  # 拡張子決め打ち
    _save(img, image_path)
    processed_count += 1                  # グローバル状態を書き換え
    if notify:
        _send_slack_notification(...)     # 通知要否もフラグに混在
    return processed_count                # 副作用なのに件数を返す(CQS違反)
After — ImageProcessor(コマンド分割)× 専用例外 × ビジネスルール判定
"""image_processor.py — コマンド・クエリ分離 × フラグ引数の排除 × null非返却（Ch13）"""
from __future__ import annotations

import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Final, Protocol

from PIL import Image

logger = logging.getLogger(__name__)

DEFAULT_RESIZE_MAX_WIDTH: Final[int] = 1200
DEFAULT_THUMBNAIL_SIZE: Final[int] = 200
THUMBNAIL_SUFFIX: Final[str] = "_thumb"

# サムネイルが必要な商品カテゴリ(ビジネスルール。フラグではなくカテゴリで判定)
THUMBNAIL_REQUIRED_CATEGORIES: Final[frozenset[str]] = frozenset({"apparel", "shoes"})


class ImageProcessingError(Exception):
    """画像処理に関する例外の基底クラス。"""


class ImageLoadError(ImageProcessingError):
    """画像の読み込みに失敗した場合の例外(Noneではなく例外で失敗を表現)。"""
    def __init__(self, path: Path, cause: Exception) -> None:
        super().__init__(f"failed to load image: {path}")
        self.path = path
        self.cause = cause


class Notifier(Protocol):
    """通知手段のプロトコル(呼び出し側が要否を判断して明示的に呼ぶ)。"""
    def send(self, message: str) -> None: ...


@dataclass(frozen=True, slots=True)
class ProcessedImage:
    """処理結果を表す値オブジェクト(クエリの戻り値。副作用を持たない)。"""
    main_path: Path
    thumbnail_path: Path | None = None


class ImageProcessor:
    """商品画像1件の加工を担うコマンドオブジェクト。

    各メソッドはコマンド(副作用あり・戻り値は最小限)とクエリ(副作用なし)を
    明確に分離し、フラグ引数による内部分岐を排除する(Ch13)。
    """

    def __init__(self, image_path: Path) -> None:
        self._image_path = image_path
        self._image = self._load(image_path)  # 完全コンストラクタ(Ch3)

    def _load(self, path: Path) -> Image.Image:
        try:
            return Image.open(path)
        except OSError as exc:
            raise ImageLoadError(path, exc) from exc

    def resize(self, max_width: int = DEFAULT_RESIZE_MAX_WIDTH) -> None:
        """画像をリサイズする(コマンド)。"""
        ratio = min(1.0, max_width / self._image.width)
        new_size = (int(self._image.width * ratio), int(self._image.height * ratio))
        self._image = self._image.resize(new_size)

    def apply_watermark(self) -> None:
        """透かしを合成する(コマンド)。"""
        self._image = _apply_watermark(self._image)

    def save(self) -> Path:
        """加工済み画像を保存する(コマンド。保存先パスのみ返す)。"""
        self._image.save(self._image_path)
        return self._image_path

    def save_thumbnail(self, size: int = DEFAULT_THUMBNAIL_SIZE) -> Path:
        """サムネイルを生成して保存する(コマンド。拡張子に依存しないwith_stemでパス生成)。"""
        thumb = self._image.copy()
        thumb.thumbnail((size, size))
        thumb_path = self._image_path.with_stem(f"{self._image_path.stem}{THUMBNAIL_SUFFIX}")
        thumb.save(thumb_path)
        return thumb_path


def handle_upload_event(image_path: Path, category: str, notifier: Notifier) -> ProcessedImage:
    """Cloud Storage アップロードイベント(Eventarc経由)を処理する。

    Raises:
        ImageLoadError: 画像の読み込みに失敗した場合(Noneは返さない)。
    """
    processor = ImageProcessor(image_path)
    processor.resize()
    processor.apply_watermark()
    main_path = processor.save()

    # フラグ引数ではなく、ビジネスルール(カテゴリ)に基づき呼び出し側がメソッドを選ぶ
    thumbnail_path = (
        processor.save_thumbnail() if category in THUMBNAIL_REQUIRED_CATEGORIES else None
    )

    notifier.send(f"{main_path} processed")  # 通知の要否は呼び出し側の責務(Ch6-7)
    return ProcessedImage(main_path=main_path, thumbnail_path=thumbnail_path)
class FakeNotifier:
    def send(self, message: str) -> None:
        print(f"[slack] {message}")

# 正常系: apparelカテゴリはサムネイル付き
result = handle_upload_event(Path("products/shirt_001.jpg"), "apparel", FakeNotifier())
# → [slack] products/shirt_001.jpg processed
# → ProcessedImage(main_path=PosixPath('products/shirt_001.jpg'), thumbnail_path=PosixPath('products/shirt_001_thumb.jpg'))

# 正常系: foodカテゴリはサムネイル不要(フラグではなくカテゴリ判定)
result = handle_upload_event(Path("products/coffee_002.png"), "food", FakeNotifier())
# → [slack] products/coffee_002.png processed
# → ProcessedImage(main_path=PosixPath('products/coffee_002.png'), thumbnail_path=None)

# 異常系: 破損画像はNoneではなく例外を送出(呼び出し側でtry/exceptを強制)
handle_upload_event(Path("products/corrupt.jpg"), "apparel", FakeNotifier())
# → ImageLoadError: failed to load image: products/corrupt.jpg
ポイント適用した設計原則/パターン書籍対応章
resize/watermark/thumbnail/notify フラグを排除し、呼び出し側がメソッドを選択呼び出しフラグ引数の排除Ch13
「何もしない」と「失敗」のNone二重使いを解消、専用例外ImageLoadErrorに統一null返却の禁止・フェイルファストCh13
save()/save_thumbnail()は副作用+最小限の戻り値のみ、無関係な値は返さないコマンド・クエリ分離(CQS)Ch13
グローバル変数processed_countを排除、状態はImageProcessorに閉じ込めるカプセル化・完全コンストラクタCh3
Notifierを呼び出し側が注入、通知要否の判断も呼び出し側に委譲単一責任原則・依存性注入Ch6-7
 
# tests/test_image_processor.py
import pytest
from pathlib import Path
from image_processor import (
    ImageProcessor, ImageLoadError, handle_upload_event, ProcessedImage,
)


class FakeNotifier:
    def __init__(self): self.sent = []
    def send(self, message): self.sent.append(message)


class TestHandleUploadEvent:
    def test_apparel_category_gets_thumbnail(self, tmp_path, sample_jpeg):
        notifier = FakeNotifier()
        result = handle_upload_event(sample_jpeg, "apparel", notifier)
        assert isinstance(result, ProcessedImage)
        assert result.thumbnail_path is not None
        assert notifier.sent == [f"{result.main_path} processed"]

    def test_food_category_skips_thumbnail(self, sample_jpeg):
        result = handle_upload_event(sample_jpeg, "food", FakeNotifier())
        assert result.thumbnail_path is None

    def test_corrupt_image_raises_instead_of_none(self, tmp_path):
        corrupt = tmp_path / "corrupt.jpg"
        corrupt.write_bytes(b"not an image")
        with pytest.raises(ImageLoadError):
            handle_upload_event(corrupt, "apparel", FakeNotifier())


class TestImageProcessorSave:
    def test_save_returns_path_only(self, sample_jpeg):
        processor = ImageProcessor(sample_jpeg)
        processor.resize()
        result = processor.save()
        assert result == sample_jpeg  # 保存先パスのみ(件数など無関係な値を返さない)

問題 B: インフラ — Cloud Run + Eventarc 商品画像処理ワーカーの冪等性 × 最小権限 × デッドレターキュー

問題Aの画像処理ハンドラは、Cloud Storage への商品画像アップロードを Eventarc が検知し、Cloud Run サービスを invoke する構成で稼働しています。現状の構成には以下の課題があります。

現状の課題:
  • Eventarc（Cloud Storage finalize イベント）はat-least-once配信を保証するため同一イベントが複数回届くことがあるが、ハンドラ側に冪等性チェックがなく、同じ画像が複数回処理され、BigQueryの処理ログテーブルに重複行が挿入され、Slack通知も複数回飛ぶ
  • Cloud Runのサービスアカウントに roles/editor（プロジェクト全体の編集者権限）が付与されており、画像バケットとBigQueryデータセット以外にも広範な操作が可能な状態になっている
  • Cloud Runの concurrency がデフォルト値80のまま、resources も cpu: 1 / memory: 256Mi の初期設定のまま変更されていない → Pillowでの画像処理はメモリを消費するため、同時実行数が増えるとOOMKilledが頻発し、Eventarcの自動リトライでさらに負荷が増す悪循環になっている
  • Eventarc triggerに再試行ポリシー（最大試行回数・バックオフ）やデッドレターキューが設定されておらず、壊れた画像ファイル（処理不能なpoison message）が無限にリトライされ続け、Cloud Runの実行コストが積み上がっている
  • Cloud Runサービスが allUsers に対して公開（ingress: all）されており、Eventarc以外からも直接HTTPリクエストで内部処理エンドポイントを叩ける状態になっている

要件

#要件
1イベントID（CloudEventの ce-id）をキーにした冪等性チェックを実装し、重複配信を無視すること
2Cloud RunのサービスアカウントをTerraformでバケット単位／データセット単位の最小権限に絞ること
3concurrencyとresourcesを画像処理の実測メモリ使用量に基づいて適正化すること（画像処理はCPU/メモリバウンドで並列度の恩恵が薄いためconcurrency=1程度が目安）
4Eventarcの配信経路（Pub/Sub）に再試行ポリシーとデッドレター用トピックを設定すること
5Cloud Runのingressを制限し、Eventarcのサービスエージェントのみが呼び出せるようにすること（allUsers剥奪）
期待する回答形式: Python（冪等性チェック実装）+ Terraform（Cloud Run/Eventarc/IAM）+ Bad vs Good 比較表 + 確認コマンド

ヒント B（段階的開示）

ヒント1 — 方向性
Eventarcは Pub/Sub をベースにしており、Pub/Subの配信保証はat-least-once（最低1回、重複ありうる）。冪等性を持たないハンドラをこの上に置くと、必ずいつか重複処理が発生する。Cloud Runの権限設計は「誰が呼べるか（ingress/invoker）」と「呼ばれた後に何ができるか（サービスアカウントの権限）」の2軸で最小化する。リソースサイジングは「CPUバウンドかメモリバウンドか」というワークロード特性で決める。
ヒント2 — アプローチ
  • BigQueryの処理ログテーブルに event_id をユニークキーとして持たせ、SELECT ... WHERE event_id = @event_id で既処理判定してから処理する
  • google_service_account + google_storage_bucket_iam_member（バケット単位）+ google_bigquery_dataset_iam_member（データセット単位）で最小権限化
  • Cloud Run v2の max_instance_request_concurrency = 1、resources.limitsを実測ピークの1.2〜1.5倍に設定
  • Eventarcの裏側にあるPub/Subサブスクリプションに dead_letter_policy（max_delivery_attempts）と retry_policy を設定
  • google_cloud_run_v2_service_iam_memberでinvokerをEventarcのP4SA（service-PROJECT_NUMBER@gcp-sa-eventarc.iam.gserviceaccount.com）のみに限定し、ingressをINGRESS_TRAFFIC_INTERNAL_ONLYにする
ヒント3 — リソースの骨格
Cloud Run v2 適正サイジング + ingress制限
resource "google_cloud_run_v2_service" "image_worker" {
  ingress = "INGRESS_TRAFFIC_INTERNAL_ONLY"
  template {
    service_account                  = google_service_account.image_worker.email
    max_instance_request_concurrency = 1
    containers {
      resources { limits = { cpu = "2", memory = "1Gi" } }
    }
  }
}
Pub/Sub デッドレター
resource "google_pubsub_subscription" "image_worker_events" {
  dead_letter_policy {
    dead_letter_topic     = google_pubsub_topic.image_worker_dlq.id
    max_delivery_attempts = 5
  }
  retry_policy {
    minimum_backoff = "10s"
    maximum_backoff = "300s"
  }
}

アーキテクチャ図 — 冪等性 × 最小権限 × デッドレターキュー

Cloud Storage bucket: product-images event: object.finalized 商品画像アップロード Eventarc Trigger product-image-upload transport: Pub/Sub(自動生成) at-least-once配信(重複あり) Pub/Sub Subscription 修正4: dead_letter_policy max_delivery_attempts=5 backoff 10s〜300s image-worker-dlq Pub/Sub Topic(DLQ) poison messageを退避 人手でトリアージ Cloud Run v2: image-worker アプリコンテナ(問題Aの成果物) 修正1: event_idで冪等性チェック (BigQuery既処理判定→重複はスキップ) 修正3: concurrency=1 resources: cpu=2 / memory=1Gi ImageProcessor.resize/apply_watermark /save/save_thumbnail アクセス制御 Before: ingress=all(allUsers公開) 修正5: ingress=INTERNAL_ONLY invoker=EventarcのP4SAのみ Before: roles/editor(プロジェクト全体) 修正2: バケット単位storage.objectAdmin + データセット単位bigquery.dataEditor OOMKilled対策(Before→After) Before: concurrency=80(既定) + cpu=1/memory=256Mi → 画像処理でメモリ枯渇 After: concurrency=1(メモリバウンドは並列化の恩恵薄い) + 実測ベースcpu=2/memory=1Gi OOMKilledによるEventarc再送の悪循環を解消 下流 BigQuery: product_image_processed_log event_id UNIQUE判定 Cloud Storage: 加工済み画像+サムネイル Slack: notifier.send() 重複時は送信されない ✓ event_idベースの冪等性チェックで重複処理・重複通知を防止 ✓ roles/editor廃止、バケット/データセット単位の最小権限に ✓ concurrency=1 + 実測ベースのリソースでOOMKilledを解消 ✓ デッドレターキューでpoison messageの無限リトライを停止 ✓ ingress制限でEventarc以外からの直接呼び出しを遮断 5回失敗でDLQへ invoke

模範解答 B

"""upload_event_handler.py — Eventarc 冪等性チェック付きハンドラ"""
from __future__ import annotations

import datetime as dt
import logging
from typing import Final

from google.cloud import bigquery

logger = logging.getLogger(__name__)

PROCESSED_LOG_TABLE: Final[str] = "mops.product_image_processed_log"


def _is_already_processed(client: bigquery.Client, event_id: str) -> bool:
    """クエリ: 副作用を持たず、既処理かどうかだけを判定する。"""
    query = f"""
        SELECT 1
        FROM `{PROCESSED_LOG_TABLE}`
        WHERE event_id = @event_id
        LIMIT 1
    """
    job = client.query(
        query,
        job_config=bigquery.QueryJobConfig(
            query_parameters=[bigquery.ScalarQueryParameter("event_id", "STRING", event_id)]
        ),
    )
    return job.result().total_rows > 0


def _record_processed(client: bigquery.Client, event_id: str, image_path: str) -> None:
    """コマンド: 処理済みイベントを記録する(副作用のみ、戻り値なし)。"""
    client.insert_rows_json(
        PROCESSED_LOG_TABLE,
        [{
            "event_id": event_id,
            "image_path": image_path,
            "processed_at": dt.datetime.now(dt.timezone.utc).isoformat(),
        }],
    )


def handle_cloud_event(event_id: str, image_path, category: str, bq_client, notifier) -> None:
    """Eventarc から届いた CloudEvent を冪等に処理する。

    Note:
        Eventarc(裏側はPub/Sub)はat-least-once配信のため、同一event_idが
        複数回届き得る。既処理であれば何もせず正常終了する(re-raiseせず
        200 OKで応答しEventarc側の再送を止める)。
    """
    if _is_already_processed(bq_client, event_id):
        logger.info("duplicate event ignored", extra={"event_id": event_id})
        return  # 冪等性: 二重配信は無視して正常終了(200)扱いにする

    handle_upload_event(image_path, category, notifier)  # 問題Aで実装したハンドラ
    _record_processed(bq_client, event_id, str(image_path))
# --- サービスアカウント: 最小権限 ---
resource "google_service_account" "image_worker" {
  account_id   = "image-worker"
  display_name = "Product Image Processor"
}

resource "google_storage_bucket_iam_member" "image_worker_bucket_rw" {
  bucket = google_storage_bucket.product_images.name
  role   = "roles/storage.objectAdmin"       # 修正2: バケット単位のみ(roles/editor全体付与を廃止)
  member = "serviceAccount:${google_service_account.image_worker.email}"
}

resource "google_bigquery_dataset_iam_member" "image_worker_bq" {
  dataset_id = google_bigquery_dataset.mops.dataset_id
  role       = "roles/bigquery.dataEditor"   # 修正2: データセット単位のみ
  member     = "serviceAccount:${google_service_account.image_worker.email}"
}

# --- Cloud Run v2: 適正サイジング + ingress制限 ---
resource "google_cloud_run_v2_service" "image_worker" {
  name     = "image-worker"
  location = "asia-northeast1"
  ingress  = "INGRESS_TRAFFIC_INTERNAL_ONLY"   # 修正5: allUsers公開を廃止

  template {
    service_account                  = google_service_account.image_worker.email
    max_instance_request_concurrency = 1        # 修正3: メモリバウンドで並列化の恩恵が薄い

    containers {
      image = "asia-northeast1-docker.pkg.dev/PROJECT_ID/mops/image-worker@sha256:def456"
      resources {
        limits = {
          cpu    = "2"    # 修正3: 実測ピークベース(旧: 1、OOMKilled頻発)
          memory = "1Gi"  # 修正3: 実測ピークベース(旧: 256Mi)
        }
      }
    }
  }
}

# Eventarcの呼び出し元(P4SA)のみをinvokerに許可(allUsersバインディングは削除)
resource "google_cloud_run_v2_service_iam_member" "eventarc_invoker" {
  name     = google_cloud_run_v2_service.image_worker.name
  location = google_cloud_run_v2_service.image_worker.location
  role     = "roles/run.invoker"
  member   = "serviceAccount:service-${var.project_number}@gcp-sa-eventarc.iam.gserviceaccount.com"
}

# --- Eventarc trigger + デッドレター(Pub/Sub) ---
resource "google_pubsub_topic" "image_worker_dlq" {
  name = "image-worker-dlq"
}

resource "google_eventarc_trigger" "image_upload" {
  name     = "product-image-upload"
  location = "asia-northeast1"

  matching_criteria {
    attribute = "type"
    value     = "google.cloud.storage.object.v1.finalized"
  }
  matching_criteria {
    attribute = "bucket"
    value     = google_storage_bucket.product_images.name
  }

  destination {
    cloud_run_service {
      service = google_cloud_run_v2_service.image_worker.name
      region  = "asia-northeast1"
    }
  }

  service_account = google_service_account.image_worker.email
}

resource "google_pubsub_subscription" "image_worker_events" {
  name  = "image-worker-events-sub"
  topic = google_eventarc_trigger.image_upload.transport[0].pubsub[0].topic

  dead_letter_policy {                       # 修正4: poison messageの無限リトライを防止
    dead_letter_topic     = google_pubsub_topic.image_worker_dlq.id
    max_delivery_attempts = 5
  }
  retry_policy {
    minimum_backoff = "10s"
    maximum_backoff = "300s"
  }
}

Bad vs Good 設計比較

観点Bad（現状）Good（改善後）
冪等性イベントID未チェック、再配信で重複処理・重複通知event_idユニーク判定でdedupe、重複は正常終了(200)扱い
IAMroles/editor(プロジェクト全体の編集者権限)バケット単位storage.objectAdmin + データセット単位bigquery.dataEditor
リソースconcurrency=80既定・cpu=1/memory=256Mi、OOMKilled頻発concurrency=1・cpu=2/memory=1Gi(実測ベース)
再試行/デッドレターポリシーなし、poison messageが無限リトライmax_delivery_attempts=5+DLQトピック、バックオフ設定
公開範囲allUsers、直接HTTPで到達可能ingress=internal-only、invokerはEventarc P4SAのみ

確認コマンド

# 1. Cloud Run のingress設定確認
gcloud run services describe image-worker --region asia-northeast1 \
  --format="value(spec.template.metadata.annotations['run.googleapis.com/ingress'])"
# Expected: internal

# 2. サービスアカウントの権限確認(roles/editorが付与されていないこと)
gcloud projects get-iam-policy PROJECT_ID --flatten="bindings[].members" \
  --filter="bindings.members:image-worker@PROJECT_ID.iam.gserviceaccount.com"
# Expected: roles/storage.objectAdmin(バケット単位)とroles/bigquery.dataEditor(データセット単位)のみ

# 3. デッドレター設定の確認
gcloud pubsub subscriptions describe image-worker-events-sub --format="yaml(deadLetterPolicy)"
# Expected: deadLetterTopic=image-worker-dlq, maxDeliveryAttempts=5

# 4. 冪等性の疎通確認(同一event_idを2回送信)
curl -X POST $CLOUD_RUN_URL -H "ce-id: test-event-001" -d '{"name":"products/test.jpg","category":"apparel"}'
curl -X POST $CLOUD_RUN_URL -H "ce-id: test-event-001" -d '{"name":"products/test.jpg","category":"apparel"}'
gcloud logging read 'resource.type="cloud_run_revision" jsonPayload.event_id="test-event-001"' --limit 5
# Expected: 2回目のリクエストで "duplicate event ignored" のログが出力されること

# 5. リソース設定の確認(OOMKilled解消)
gcloud run services describe image-worker --region asia-northeast1 \
  --format="value(spec.template.spec.containers[0].resources.limits)"
# Expected: cpu=2, memory=1Gi

ポイント解説

カテゴリ A

1 フラグ引数は「内部にif文を隠す」アンチパターン（Ch13）
resize/watermark/thumbnail/notify の4フラグは実質 2^4=16通り の内部分岐を1関数に押し込めており、実際に使われる組み合わせをコードから読み取れない。ImageProcessor のようにメソッドを分割し、呼び出し側（ビジネスルール）が明示的に呼ぶ設計にすれば、使われている組み合わせがそのままコードとして可視化される。
2 nullは「失敗」と「意図的に何もしない」を区別できない（Ch13）
元のコードは両方のケースで None を返しており、呼び出し側は一律「error」として扱うしかなかった。専用例外（ImageLoadError）に統一することで、呼び出し側は try/except で明示的にハンドリングできる。
3 コマンド・クエリ分離（CQS）を守ると呼び出し側の意図が明確になる（Ch13）
save()/save_thumbnail() は「保存する」という副作用（コマンド）の結果として保存先パスだけを返す。元のコードのように無関係な処理件数を返すと、呼び出し側は「これは何の値か」を都度コードを読んで確認する必要があった。

カテゴリ B

4 at-least-once配信を前提にした冪等性はイベント駆動アーキテクチャの必須要件
Eventarcの裏側はPub/Subであり、配信保証は「最低1回」。ハンドラ側で event_id ベースのdedupeを行わない限り、ネットワーク遅延やCloud Run側の再起動など様々な要因で重複処理が発生しうる。
5 Cloud Runの権限設計は「誰が呼べるか」と「何ができるか」の2軸で最小化する
ingress/invokerで呼び出し元を制限し、サービスアカウントの権限をバケット・データセット単位に絞ることで、どちらか一方が破られても被害範囲を限定できる（多層防御）。
6 デッドレターキューは「無限リトライによるコスト垂れ流し」を止める安全弁
壊れた画像のような恒久的に失敗するイベントは、何度リトライしても成功しない。max_delivery_attempts で見切りをつけてDLQに退避し、人間が後からトリアージできるようにする。

実務への応用

  • フラグ引数の排除パターンは、MOpsのバッチ処理全般（クーポン発行・通知配信・レポート生成等）で「オプションを増やしすぎた関数」を見直す際にそのまま適用できる: 「このフラグは本当に呼び出し側の関心事か、それとも内部実装の都合か」を切り分けることが第一歩
  • コマンド・クエリ分離は、Pull Request レビューで「この関数の戻り値は何のために存在するか」を問うだけで機械的にチェックできる: 副作用のある関数が予期しない値を返していたら、それはCQS違反のサインであることが多い
  • Eventarc + Cloud Runのようなイベント駆動構成は、画像処理だけでなく注文確定メール送信・在庫同期・レビュー投稿の不正検知など、Cloud Storage/Pub/Subをトリガーとするあらゆる非同期処理に横展開できる標準パターン: 「配信保証はat-least-once」を前提に冪等性を最初から設計に組み込むことが、後からのバグ調査コストを大きく下げる
  • Cloud Runの最小権限化（roles/editor廃止 → リソース単位の権限）は、監査対応・セキュリティレビューで頻出する指摘事項をあらかじめ潰しておける即効性の高い改善

今日のまとめ

カテゴリAでは、4つのフラグ引数で内部分岐を隠し、失敗と「何もしない」を同じ None で表現し、副作用と無関係な戻り値を混同していた process_image を、ImageProcessor のコマンドメソッド群とビジネスルールに基づく呼び出し側の明示的な選択に置き換え、コマンド・クエリ分離とnull非返却を徹底した（Ch13、Ch3/Ch6-7）。

カテゴリBでは、at-least-once配信を前提にした冪等性チェックをハンドラに組み込み、roles/editorという過剰権限をリソース単位の最小権限に、allUsers公開をEventarc限定のingressに、無限リトライをデッドレターキューに、それぞれ置き換えた。どちらも共通するのは「暗黙のうちに許してしまっていた不確実性（フラグの組み合わせ・重複配信・過剰な信頼範囲）を、明示的な設計判断に置き換える」という設計思想である。

次のステップ

  • 発展問題: ImageProcessor に「透かし画像の種類（セール告知/ブランドロゴ）」を追加する際、フラグ引数を増やさずに拡張する方法を検討する（ストラテジパターン or 値オブジェクトでの表現、Ch8/Ch5）
  • 発展問題: デッドレターキュー（image-worker-dlq）に溜まったpoison messageを定期的に検知しSlack通知するCloud Schedulerジョブを設計する（2026-07-18のグレースフルシャットダウン設計と組み合わせ、監視・アラート設計へ発展させる）
  • 参考: 「良いコード・悪いコードで学ぶ設計入門」Ch3（カプセル化）/ Ch6-7（関心の分離）/ Ch10（設計の悪魔）/ Ch13（メソッド設計）、Eventarc（Cloud Storage trigger）、Cloud Run v2 ingress設定、Pub/Sub dead letter policy

自己評価（あとで記入）

自分の回答

気づき・メモ