✍️ 필사 모드: 플랫폼 아키텍처 의사결정 가이드: 모놀리스와 마이크로서비스 사이

[모놀리스] -----> [모듈러 모놀리스] -----> [매크로서비스] -----> [마이크로서비스]

  단일 배포 단위      모듈 경계 분리        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 인증 범위를 축소하기 위해 결제 처리를 별도 서비스로 분리해야 한다.
또한 결제 서비스는 주문 서비스와 독립적으로 스케일링해야 한다
(Black Friday 기간 결제 트래픽이 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)로 결제 분리: cold start 이슈로 p99 지연 불확실

Phase 1: 모듈러 모놀리스화 (1-3개월)
  - 코드 내 모듈 경계 정립
  - facade 인터페이스 도입
  - 경계 위반 감지 CI 구축
  - DB 스키마를 모듈별로 논리 분리

Phase 2: 첫 번째 서비스 추출 (2-4개월)
  - 가장 독립적인 모듈 1개를 서비스로 분리
  - 통신 프로토콜 결정 (gRPC/REST/이벤트)
  - Saga 또는 이벤트 기반 일관성 구현
  - A/B 라우팅으로 신규 서비스와 기존 모듈 병행 운영

Phase 3: 안정화 및 확장 (3-6개월)
  - 첫 번째 서비스의 운영 안정성 확인
  - 모니터링, 알림, 런북 정비
  - 다음 분리 대상 선정 및 반복

Phase 4: 플랫폼 성숙 (지속적)
  - service mesh / API gateway 도입
  - 분산 트레이싱 체계 구축
  - 서비스 카탈로그 및 ownership 관리

"""
마이그레이션 단계 진행 가능 여부를 판단하는 게이트 체크.

각 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"