A+B 複合 — ループネスト解消 × コレクションのカプセル化（注文キャンセル明細処理 Bad→Good 第9章）× order-cancellation-service Pub/Sub Push配信の信頼性設計（デッドレターキュー × Exactly-once × 順序保証 × OIDC検証）

📋 コピー# tests/test_cancelled_items.py
import pytest
from cancelled_items import CancelledItem, CancelledItems


class TestCancelledItemsOf:
    def test_deduplicates_same_sku(self):
        items = [CancelledItem("SKU-A", 1000, 1), CancelledItem("SKU-A", 1000, 1)]
        assert len(CancelledItems.of(items)) == 1


class TestOverlappingWith:
    def test_returns_only_matching_skus(self):
        items = [CancelledItem("SKU-A", 1000, 1), CancelledItem("SKU-B", 1000, 1)]
        cancelled = CancelledItems.of(items)
        overlap = cancelled.overlapping_with(frozenset({"SKU-A"}))
        assert [i.sku for i in overlap] == ["SKU-A"]

    def test_does_not_mutate_original(self):
        items = [CancelledItem("SKU-A", 1000, 1)]
        cancelled = CancelledItems.of(items)
        cancelled.overlapping_with(frozenset({"SKU-A"}))
        assert len(cancelled) == 1  # 元のコレクションは不変のまま


class TestCountAbovePrice:
    def test_counts_only_above_threshold(self):
        items = [CancelledItem("SKU-A", 12_000, 1), CancelledItem("SKU-B", 3_000, 1)]
        assert CancelledItems.of(items).count_above_price() == 1

📋 コピー"""order_cancellation_push_handler.py — OIDC検証(修正5) × 冪等性チェック(修正4)"""
from __future__ import annotations

import base64
import json
from dataclasses import dataclass
from typing import Final

from fastapi import FastAPI, HTTPException, Request, Response
from google.auth.transport import requests as google_requests
from google.oauth2 import id_token

app = FastAPI()

_EXPECTED_AUDIENCE: Final[str] = "https://order-cancellation-service-xxxx.a.run.app"
_EXPECTED_SERVICE_ACCOUNT: Final[str] = "pubsub-pusher@mops-prod.iam.gserviceaccount.com"
_IDEMPOTENCY_TTL_SECONDS: Final[int] = 86_400


@dataclass(frozen=True, slots=True)
class CancellationEvent:
    """order-eventsトピックから受信するキャンセルイベント(不変)。"""
    order_id: str
    message_id: str


def _verify_pubsub_oidc_token(request: Request) -> None:
    """Pub/Subからの正規リクエストであることをOIDCトークンで検証する(修正5)。

    Args:
        request: Cloud Runが受け取ったHTTPリクエスト。

    Raises:
        HTTPException: トークンが無効、またはaudience/service accountが不一致の場合。
    """
    auth_header = request.headers.get("authorization", "")
    if not auth_header.startswith("Bearer "):
        raise HTTPException(status_code=403, detail="missing bearer token")

    token = auth_header.removeprefix("Bearer ").strip()
    try:
        claims = id_token.verify_oauth2_token(
            token, google_requests.Request(), audience=_EXPECTED_AUDIENCE
        )
    except ValueError as exc:
        raise HTTPException(status_code=403, detail="invalid oidc token") from exc

    if claims.get("email") != _EXPECTED_SERVICE_ACCOUNT:
        raise HTTPException(status_code=403, detail="unexpected service account")


def _parse_event(envelope: dict) -> CancellationEvent:
    """Pub/Sub Pushのエンベロープからキャンセルイベントを取り出す。"""
    message = envelope["message"]
    payload = json.loads(base64.b64decode(message["data"]))
    return CancellationEvent(order_id=payload["order_id"], message_id=message["messageId"])


@app.post("/pubsub/push")
async def handle_push(request: Request) -> Response:
    """Pub/Sub Pushエンドポイント(OIDC検証 + 冪等性チェック付き)。

    Returns:
        処理成功時は200(Ack)。処理失敗時は5xx(Pub/Subが自動リトライ)。
    """
    _verify_pubsub_oidc_token(request)  # 修正5

    envelope = await request.json()
    event = _parse_event(envelope)

    if not idempotency_store.mark_if_new(event.message_id, ttl=_IDEMPOTENCY_TTL_SECONDS):  # 修正4
        # 既に処理済み(再配信) → 200を返してAckし、これ以上のリトライを止める
        return Response(status_code=200)

    process_cancellation(event.order_id)  # 在庫解放・クーポン失効・通知メール送信
    return Response(status_code=200)

Bad vs Good 設計比較

確認コマンド

📋 コピー# 1. デッドレターポリシーが設定されていることを確認
gcloud pubsub subscriptions describe order-cancelled-sub \
  --format="value(deadLetterPolicy.deadLetterTopic,deadLetterPolicy.maxDeliveryAttempts)"
# Expected: projects/mops-prod/topics/order-cancelled-dlq  5

# 2. メッセージ順序保証が有効なことを確認
gcloud pubsub subscriptions describe order-cancelled-sub \
  --format="value(enableMessageOrdering)"
# Expected: True

# 3. 冪等性チェックにより同一message_idの二重処理がスキップされることを確認
python -c "
from order_cancellation_push_handler import idempotency_store
print(idempotency_store.mark_if_new('msg-001'))  # Expected: True (初回)
print(idempotency_store.mark_if_new('msg-001'))  # Expected: False (2回目はスキップ)
"

# 4. OIDCトークンなしのリクエストが403で拒否されることを確認
curl -s -o /dev/null -w "%{http_code}\n" -X POST \
  https://order-cancellation-service-xxxx.a.run.app/pubsub/push \
  -H "Content-Type: application/json" -d '{"message":{"data":"e30=","messageId":"1"}}'
# Expected: 403

# 5. デッドレターキューの滞留アラートが設定されていることを確認
gcloud alpha monitoring policies list --filter="displayName:order-cancelled-dlq" \
  --format="value(displayName,enabled)"
# Expected: order-cancelled-dlq unacked message count  True