プラットフォームアーキテクチャ意思決定ガイド：モノリスとマイクロサービスの間

[モノリス] -----> [モジュラーモノリス] -----> [マクロサービス] -----> [マイクロサービス]

  単一デプロイ単位      モジュール境界分離        2-5個の大きなサービス      ドメイン別独立サービス
  単一DB               モジュール別スキーマ分離   サービス別DB               サービス別DB
  チーム1個             チーム1-2個              チーム2-5個               チーム5個以上
  運用シンプル           運用シンプル             運用普通                   運用複雑

"""
モジュラーモノリスのモジュール間通信例。

orderモジュールがpaymentモジュールにアクセスする際、
paymentの内部実装ではなくpublic facadeを通じる。
"""

# === payment/facade.py (paymentモジュールの公開インターフェース) ===
from dataclasses import dataclass
from typing import Optional


@dataclass
class PaymentResult:
    reservation_id: str
    status: str
    amount: int
    currency: str


class PaymentFacade:
    """Paymentモジュールの公開インターフェース。

    他のモジュールはこのクラスを通じてのみpayment機能にアクセスする。
    内部実装（repository、service、domain model）への直接アクセスは禁止。
    """

    def __init__(self, payment_service):
        self._service = payment_service

    def reserve_payment(
        self,
        customer_id: str,
        amount: int,
        currency: str = "KRW",
        idempotency_key: Optional[str] = None,
    ) -> PaymentResult:
        """決済を予約する。実際の課金はconfirm時に発生。"""
        reservation = self._service.create_reservation(
            customer_id=customer_id,
            amount=amount,
            currency=currency,
            idempotency_key=idempotency_key,
        )
        return PaymentResult(
            reservation_id=reservation.id,
            status=reservation.status.value,
            amount=reservation.amount,
            currency=reservation.currency,
        )

    def confirm_payment(self, reservation_id: str) -> PaymentResult:
        """予約された決済を確定する。"""
        result = self._service.confirm(reservation_id)
        return PaymentResult(
            reservation_id=result.id,
            status=result.status.value,
            amount=result.amount,
            currency=result.currency,
        )

    def cancel_payment(self, reservation_id: str) -> None:
        """予約された決済をキャンセルする。"""
        self._service.cancel(reservation_id)


# === order/service.py (orderモジュールでpayment facade使用) ===

class OrderService:
    """注文サービス。

    paymentモジュールの内部実装に依存せず、
    PaymentFacadeを通じてのみ決済機能にアクセスする。
    """

    def __init__(self, order_repo, payment_facade: PaymentFacade):
        self.order_repo = order_repo
        self.payment = payment_facade

    def create_order(self, customer_id: str, items: list, total: int) -> dict:
        # 1. 注文作成
        order = self.order_repo.create(
            customer_id=customer_id,
            items=items,
            total=total,
        )

        # 2. 決済予約（payment facadeを通じて）
        try:
            payment_result = self.payment.reserve_payment(
                customer_id=customer_id,
                amount=total,
                idempotency_key=f"order-{order.id}",
            )
            order.payment_reservation_id = payment_result.reservation_id
            self.order_repo.save(order)
        except Exception as e:
            order.status = "payment_failed"
            self.order_repo.save(order)
            raise

        return {"order_id": order.id, "status": order.status}

"""
モジュール境界違反検出スクリプト。

各モジュールは他モジュールのfacadeのみimportできる。
内部パッケージ（service、repository、domain）を直接importすると違反。
"""
import ast
import sys
from pathlib import Path
from dataclasses import dataclass


@dataclass
class Violation:
    file: str
    line: int
    importing_module: str
    imported_module: str
    imported_path: str
    reason: str


# モジュール一覧と許可された公開パッケージ
MODULE_CONFIG = {
    "order": {"public": ["order.facade"]},
    "payment": {"public": ["payment.facade"]},
    "inventory": {"public": ["inventory.facade"]},
    "shipping": {"public": ["shipping.facade"]},
    "user": {"public": ["user.facade"]},
}


def get_module_name(file_path: str) -> str | None:
    """ファイルパスからモジュール名を抽出する。"""
    parts = Path(file_path).parts
    for module_name in MODULE_CONFIG:
        if module_name in parts:
            return module_name
    return None


def check_file(file_path: str) -> list[Violation]:
    """単一ファイルのimportを検査し違反リストを返す。"""
    violations = []
    source_module = get_module_name(file_path)
    if source_module is None:
        return violations

    with open(file_path) as f:
        try:
            tree = ast.parse(f.read())
        except SyntaxError:
            return violations

    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                _check_import(file_path, source_module, alias.name, node.lineno, violations)
        elif isinstance(node, ast.ImportFrom) and node.module:
            _check_import(file_path, source_module, node.module, node.lineno, violations)

    return violations


def _check_import(
    file_path: str,
    source_module: str,
    imported_path: str,
    line: int,
    violations: list[Violation],
):
    """個別のimport文がモジュール境界に違反しているかチェックする。"""
    for target_module, config in MODULE_CONFIG.items():
        if target_module == source_module:
            continue  # 同一モジュール内のimportはOK

        if imported_path.startswith(target_module + "."):
            # 他モジュールをimportしている -> publicパッケージか確認
            is_public = any(
                imported_path.startswith(pub)
                for pub in config["public"]
            )
            if not is_public:
                violations.append(Violation(
                    file=file_path,
                    line=line,
                    importing_module=source_module,
                    imported_module=target_module,
                    imported_path=imported_path,
                    reason=f"Direct import of '{target_module}' internals. "
                           f"Use {config['public']} instead.",
                ))


def main():
    """プロジェクト全ファイルを検査し違反を報告する。"""
    project_root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(".")
    all_violations = []

    for py_file in project_root.rglob("*.py"):
        violations = check_file(str(py_file))
        all_violations.extend(violations)

    if all_violations:
        print(f"\n{'='*60}")
        print(f"Module boundary violations found: {len(all_violations)}")
        print(f"{'='*60}\n")
        for v in all_violations:
            print(f"  {v.file}:{v.line}")
            print(f"    {v.importing_module} -> {v.imported_module} ({v.imported_path})")
            print(f"    {v.reason}\n")
        sys.exit(1)
    else:
        print("No module boundary violations found.")
        sys.exit(0)


if __name__ == "__main__":
    main()

# .github/workflows/boundary-check.yml
name: Module Boundary Check
on: [pull_request]
jobs:
  check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - run: python scripts/check_module_boundaries.py src/

# ADR-007: 決済モジュールを独立サービスとして分離

## ステータス: 承認済み (2026-02-15)

## コンテキスト

現在、決済モジュールはモノリス内部でfacadeを通じて呼び出されている。
PCI-DSS認証範囲を縮小するため、決済処理を別サービスに分離する必要がある。
また、決済サービスは注文サービスとは独立してスケーリングする必要がある
（ブラックフライデー期間中、決済トラフィックが10倍に増加）。

## 決定

決済モジュールを独立したgRPCサービスとして分離する。

- 通信: gRPC（内部）、REST（外部PG連携）
- DB: 別のPostgreSQLインスタンス
- 整合性: Sagaパターン（orchestration方式）
- デプロイ: 独立したKubernetes deployment

## 結果

### ポジティブ

- PCI-DSS認証範囲が決済サービスに限定される
- 決済サービスの独立スケーリングが可能
- 決済ロジック変更時に注文サービスの再デプロイ不要

### ネガティブ

- サービス間通信レイテンシの追加（約5ms）
- Saga補償トランザクションの実装と運用が必要
- 運用複雑度の増加：別のモニタリング、デプロイパイプライン

### 数値根拠

- 現在の決済p99レイテンシ: 150ms -> 分離後予想: 155ms（許容範囲）
- 現在のPCI認証範囲: インフラ全体 -> 分離後: 決済サービスのみ
- インフラコスト増加: 月額$200（別DB＋サービスインスタンス）

## 検討した代替案

1. モノリス維持 + PCI範囲全体認証: コストと監査負担が大きい
2. サーバーレス（Lambda）で決済分離: コールドスタート問題によりp99レイテンシが不確実

Phase 1: モジュラーモノリス化 (1-3ヶ月)
  - コード内のモジュール境界確立
  - facadeインターフェース導入
  - 境界違反検出CI構築
  - DBスキーマをモジュール別に論理分離

Phase 2: 最初のサービス抽出 (2-4ヶ月)
  - 最も独立したモジュール1つをサービスとして分離
  - 通信プロトコル決定（gRPC/REST/イベント）
  - Sagaまたはイベントベースの整合性実装
  - A/Bルーティングで新サービスと既存モジュールの並行運用

Phase 3: 安定化と拡張 (3-6ヶ月)
  - 最初のサービスの運用安定性確認
  - モニタリング、アラート、ランブック整備
  - 次の分離対象選定と繰り返し

Phase 4: プラットフォーム成熟 (継続的)
  - サービスメッシュ / APIゲートウェイ導入
  - 分散トレーシング体系構築
  - サービスカタログとオーナーシップ管理

"""
マイグレーション段階の進行可否を判断するゲートチェック。

各phase完了後、次のphaseに進む前に
このチェックを通過する必要がある。
"""
from dataclasses import dataclass


@dataclass
class PhaseGateCheck:
    name: str
    passed: bool
    detail: str


def check_phase1_gate() -> list[PhaseGateCheck]:
    """Phase 1完了ゲート：モジュラーモノリス化が十分に行われたか。"""
    return [
        PhaseGateCheck(
            name="module_boundaries_defined",
            passed=True,  # 実際にはコード分析結果を確認
            detail="All modules have facade interfaces",
        ),
        PhaseGateCheck(
            name="no_boundary_violations",
            passed=True,  # CIで違反0件確認
            detail="0 boundary violations in last 30 days",
        ),
        PhaseGateCheck(
            name="schema_separated",
            passed=True,  # DBスキーマ分離確認
            detail="Each module uses its own schema prefix",
        ),
        PhaseGateCheck(
            name="integration_tests_exist",
            passed=True,
            detail="Module integration tests cover 85%+ of facade methods",
        ),
    ]


def check_phase2_gate() -> list[PhaseGateCheck]:
    """Phase 2完了ゲート：最初のサービス抽出が安定しているか。"""
    return [
        PhaseGateCheck(
            name="service_p99_latency",
            passed=True,   # 実際のメトリクスに基づく判断
            detail="p99 latency < 200ms for 14 consecutive days",
        ),
        PhaseGateCheck(
            name="error_rate",
            passed=True,
            detail="Error rate < 0.1% for 14 consecutive days",
        ),
        PhaseGateCheck(
            name="saga_compensation_tested",
            passed=True,
            detail="Compensation scenarios tested in staging 3+ times",
        ),
        PhaseGateCheck(
            name="runbook_documented",
            passed=True,
            detail="Incident runbook reviewed by on-call team",
        ),
        PhaseGateCheck(
            name="rollback_verified",
            passed=True,
            detail="Rollback to monolith path verified in staging",
        ),
    ]


def evaluate_gate(checks: list[PhaseGateCheck]) -> tuple[bool, str]:
    """ゲート通過可否を判断する。"""
    failed = [c for c in checks if not c.passed]
    if failed:
        details = "; ".join(f"{c.name}: {c.detail}" for c in failed)
        return False, f"Gate BLOCKED - {len(failed)} checks failed: {details}"
    return True, "Gate PASSED - all checks passed"