billing/
├── domain/                     # ноль зависимостей от фреймворков
│   ├── money.py                # Money, Percent — объекты-значения
│   ├── period.py               # BillingPeriod
│   ├── plan.py                 # Plan, PlanCode — снимок из каталога
│   ├── subscription.py         # агрегат: корень + инварианты + события
│   ├── events.py               # доменные события
│   ├── errors.py               # доменные исключения
│   └── ports.py                # Protocol: репозиторий, UoW, часы, каталог
├── application/
│   ├── commands.py             # ChangePlan, AssignSeats, CancelSubscription
│   └── handlers.py             # сценарии: 5–15 строк каждый
├── infrastructure/
│   ├── schema.sql
│   ├── pg_subscription_repo.py # маппинг агрегата на таблицы + версия
│   ├── unit_of_work.py         # транзакция + слив событий в outbox
│   └── outbox_relay.py         # фоновая доставка в брокер
├── api/
│   ├── dto.py                  # pydantic-модели запроса/ответа
│   ├── routes.py               # FastAPI
│   └── error_mapping.py        # доменное исключение → HTTP-код
└── tests/
    ├── unit/                   # домен, без моков и без БД
    ├── integration/            # репозиторий против настоящего Postgres
    └── e2e/                    # HTTP-уровень

# domain/money.py
from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP

CENTS = Decimal("0.01")


class CurrencyMismatch(ValueError):
    """Складывать рубли с долларами — не ошибка ввода, а ошибка модели."""


@dataclass(frozen=True, slots=True)
class Money:
    amount: Decimal
    currency: str

    def __post_init__(self) -> None:
        # float здесь запрещён навсегда: 0.1 + 0.2 != 0.3, и в отчётности это всплывёт
        if not isinstance(self.amount, Decimal):
            raise TypeError("сумма должна быть Decimal, а не float/str")
        if len(self.currency) != 3 or not self.currency.isupper():
            raise ValueError("валюта — код ISO-4217 из трёх заглавных букв")
        if -self.amount.as_tuple().exponent > 2:
            raise ValueError("больше двух знаков после запятой: копейки уже потеряны")

    @classmethod
    def rub(cls, amount: str) -> "Money":
        return cls(Decimal(amount), "RUB")

    @classmethod
    def zero(cls, currency: str) -> "Money":
        return cls(Decimal("0.00"), currency)

    def _same_currency(self, other: "Money") -> None:
        if self.currency != other.currency:
            raise CurrencyMismatch(f"{self.currency} и {other.currency}")

    def __add__(self, other: "Money") -> "Money":
        self._same_currency(other)
        return Money(self.amount + other.amount, self.currency)

    def __sub__(self, other: "Money") -> "Money":
        self._same_currency(other)
        return Money(self.amount - other.amount, self.currency)

    def prorate(self, used_days: int, total_days: int) -> "Money":
        """Доля суммы за used_days из total_days. Округление РОВНО ОДИН раз, в конце."""
        if total_days <= 0 or not (0 <= used_days <= total_days):
            raise ValueError("некорректная пропорция")
        exact = self.amount * used_days / total_days
        return Money(exact.quantize(CENTS, rounding=ROUND_HALF_UP), self.currency)

# domain/period.py
import calendar
from dataclasses import dataclass
from datetime import date, timedelta


def add_one_month(d: date) -> date:
    """1 янв → 1 фев; 31 янв → 28 фев (клип на последний день месяца)."""
    year, month = (d.year + 1, 1) if d.month == 12 else (d.year, d.month + 1)
    return date(year, month, min(d.day, calendar.monthrange(year, month)[1]))


@dataclass(frozen=True, slots=True)
class BillingPeriod:
    start: date
    end: date  # включительно

    def __post_init__(self) -> None:
        if self.end < self.start:
            raise ValueError("период заканчивается раньше, чем начинается")

    @property
    def total_days(self) -> int:
        return (self.end - self.start).days + 1

    def remaining_days(self, on: date) -> int:
        """Сколько дней периода ещё не использовано, считая сам день `on`."""
        if not self.start <= on <= self.end:
            raise ValueError(f"{on} вне периода {self.start}..{self.end}")
        return (self.end - on).days + 1

    def next(self) -> "BillingPeriod":
        new_start = self.end + timedelta(days=1)
        return BillingPeriod(new_start, add_one_month(new_start) - timedelta(days=1))

    @classmethod
    def monthly_from(cls, start: date) -> "BillingPeriod":
        return cls(start, add_one_month(start) - timedelta(days=1))

# domain/subscription.py
from __future__ import annotations

from datetime import date, timedelta
from enum import Enum
from typing import NewType
from uuid import UUID

from .errors import (InvalidTransition, SeatLimitExceeded, TrialAlreadyUsed,
                     SamePlan)
from .events import (PlanChanged, SubscriptionActivated, SubscriptionExpired,
                     CancellationRequested, DomainEvent)
from .money import Money
from .period import BillingPeriod
from .plan import Plan, PlanCode

SubscriptionId = NewType("SubscriptionId", UUID)
CustomerId = NewType("CustomerId", UUID)

TRIAL_DAYS = 14
GRACE_DAYS = 7


class Status(str, Enum):
    TRIALING = "trialing"
    ACTIVE = "active"
    PAST_DUE = "past_due"
    PENDING_CANCELLATION = "pending_cancellation"
    EXPIRED = "expired"


CHANGEABLE = {Status.TRIALING, Status.ACTIVE, Status.PENDING_CANCELLATION}


class Subscription:
    """Корень агрегата. Единственное место, где меняется состояние подписки."""

    def __init__(
        self,
        subscription_id: SubscriptionId, customer_id: CustomerId,
        plan_code: PlanCode, price: Money, period: BillingPeriod, status: Status,
        seats: int, max_seats: int, assigned_seats: int = 0,
        trial_used: bool = False, grace_until: date | None = None, version: int = 0,
    ) -> None:
        # приватность по соглашению: снаружи — только методы и свойства на чтение
        self._id, self._customer_id = subscription_id, customer_id
        self._plan_code, self._price = plan_code, price
        self._period, self._status, self._grace_until = period, status, grace_until
        self._seats, self._max_seats, self._assigned_seats = seats, max_seats, assigned_seats
        self._trial_used, self._version = trial_used, version
        self._events: list[DomainEvent] = []

    # ---------- фабрика ----------

    @classmethod
    def start_trial(
        cls, subscription_id: SubscriptionId, customer_id: CustomerId,
        plan: Plan, seats: int, today: date, customer_had_trial: bool,
    ) -> "Subscription":
        if customer_had_trial:
            raise TrialAlreadyUsed(f"клиент {customer_id} уже использовал пробный период")
        if not 1 <= seats <= plan.max_seats:
            raise SeatLimitExceeded(f"план {plan.code} допускает до {plan.max_seats} мест")
        return cls(
            subscription_id=subscription_id, customer_id=customer_id,
            plan_code=plan.code, price=plan.price,
            period=BillingPeriod.monthly_from(today), status=Status.TRIALING,
            seats=seats, max_seats=plan.max_seats, trial_used=True,
        )

    # ---------- чтение ----------
    # свойства нужны репозиторию для маппинга и сценарию для ответа;
    # сеттеров нет ни одного — состояние меняется только глаголами домена
    id = property(lambda self: self._id)
    status = property(lambda self: self._status)
    plan_code = property(lambda self: self._plan_code)
    price = property(lambda self: self._price)
    period = property(lambda self: self._period)
    seats = property(lambda self: self._seats)
    assigned_seats = property(lambda self: self._assigned_seats)
    max_seats = property(lambda self: self._max_seats)
    grace_until = property(lambda self: self._grace_until)
    version = property(lambda self: self._version)

    # ---------- поведение ----------

    def change_plan(self, new_plan: Plan, on: date) -> Money:
        """Меняет план и возвращает сумму к доплате (может быть отрицательной = кредит)."""
        if self._status not in CHANGEABLE:
            raise InvalidTransition(f"смена плана недоступна в статусе {self._status.value}")
        if new_plan.code == self._plan_code:
            raise SamePlan(f"подписка уже на плане {new_plan.code}")
        if self._assigned_seats > new_plan.max_seats:
            raise SeatLimitExceeded(
                f"назначено {self._assigned_seats} мест, план {new_plan.code} "
                f"допускает {new_plan.max_seats}: сначала освободите места"
            )

        remaining = self._period.remaining_days(on)
        total = self._period.total_days

        if self._status is Status.TRIALING:
            # в триале деньги ещё не списаны — пересчитывать нечего
            delta = Money.zero(new_plan.price.currency)
        else:
            credit = self._price.prorate(remaining, total)          # вернуть за старый план
            charge = new_plan.price.prorate(remaining, total)       # взять за новый
            delta = charge - credit

        old_code = self._plan_code
        self._plan_code = new_plan.code
        self._price = new_plan.price
        self._max_seats = new_plan.max_seats
        self._seats = min(self._seats, new_plan.max_seats)

        self._raise(PlanChanged(
            subscription_id=self._id, old_plan=old_code, new_plan=new_plan.code,
            delta=delta, effective_on=on,
        ))
        return delta

    def assign_seats(self, count: int) -> None:
        if count < 0:
            raise ValueError("отрицательное число мест")
        if count > self._seats:
            raise SeatLimitExceeded(f"оплачено {self._seats} мест, назначить пытаются {count}")
        self._assigned_seats = count

    def request_cancellation(self, on: date) -> None:
        if self._status not in {Status.ACTIVE, Status.TRIALING, Status.PAST_DUE}:
            raise InvalidTransition(f"нельзя отменить подписку в статусе {self._status.value}")
        self._status = Status.PENDING_CANCELLATION
        self._raise(CancellationRequested(
            subscription_id=self._id, requested_on=on, effective_on=self._period.end,
        ))

    def settle_payment(self, on: date) -> None:
        """Успешное списание: активирует триал или продлевает период."""
        if self._status is Status.EXPIRED:
            raise InvalidTransition("истёкшая подписка не продлевается — заводите новую")
        was_trial = self._status is Status.TRIALING
        self._status = Status.ACTIVE
        self._grace_until = None
        self._period = self._period.next()
        if was_trial:
            self._raise(SubscriptionActivated(subscription_id=self._id, activated_on=on))

    def mark_past_due(self, on: date) -> None:
        if self._status not in {Status.ACTIVE, Status.TRIALING}:
            raise InvalidTransition(f"просрочка неприменима к статусу {self._status.value}")
        self._status, self._grace_until = Status.PAST_DUE, on + timedelta(days=GRACE_DAYS)

    # resume() симметричен request_cancellation и опущен здесь ради краткости

    def expire(self, on: date) -> None:
        """Вызывается планировщиком: конец grace или конец периода при отмене."""
        if self._status is Status.EXPIRED:
            return  # идемпотентно: повторный вызов планировщика ничего не ломает
        allowed = (
            (self._status is Status.PAST_DUE and self._grace_until and on > self._grace_until)
            or (self._status is Status.PENDING_CANCELLATION and on > self._period.end)
            or (self._status is Status.TRIALING and on > self._period.end)
        )
        if not allowed:
            raise InvalidTransition("срок ещё не истёк — принудительное истечение запрещено")
        self._status = Status.EXPIRED
        self._raise(SubscriptionExpired(subscription_id=self._id, expired_on=on))

    # ---------- события ----------

    def _raise(self, event: DomainEvent) -> None:
        self._events.append(event)

    def pull_events(self) -> list[DomainEvent]:
        events, self._events = self._events, []
        return events

def test_upgrade_in_the_middle_of_period():
    sub = active_subscription(price=Money.rub("3000.00"),
                              period=BillingPeriod(date(2026, 11, 1), date(2026, 11, 30)))
    delta = sub.change_plan(plan("business", Money.rub("7500.00")), on=date(2026, 11, 12))
    assert delta == Money.rub("2850.00")   # 4750 − 1900

# domain/ports.py
from datetime import date
from typing import Protocol

from .plan import Plan, PlanCode
from .subscription import Subscription, SubscriptionId


class SubscriptionRepository(Protocol):
    def get(self, subscription_id: SubscriptionId) -> Subscription:
        """Возвращает агрегат целиком или бросает SubscriptionNotFound."""

    def add(self, subscription: Subscription) -> None: ...

    def save(self, subscription: Subscription) -> None:
        """Сохраняет; бросает ConcurrencyConflict, если версия устарела."""


class PlanCatalog(Protocol):
    def get(self, code: PlanCode) -> Plan: ...


class Clock(Protocol):
    def today(self) -> date: ...


class UnitOfWork(Protocol):
    subscriptions: SubscriptionRepository

    def __enter__(self) -> "UnitOfWork": ...
    def __exit__(self, *exc) -> None: ...
    def commit(self) -> None: ...
    def rollback(self) -> None: ...

# application/commands.py
from dataclasses import dataclass
from datetime import date
from uuid import UUID

from domain.money import Money


@dataclass(frozen=True, slots=True)
class ChangePlan:
    subscription_id: UUID
    new_plan_code: str
    idempotency_key: str


@dataclass(frozen=True, slots=True)
class ChangePlanResult:
    """Результат сценария — не агрегат наружу, а ровно то, что нужно вызывающему."""
    charged_now: Money
    next_billing_date: date

# application/handlers.py
from datetime import timedelta

from domain.plan import PlanCode
from domain.ports import Clock, PlanCatalog, UnitOfWork
from domain.subscription import SubscriptionId

from .commands import ChangePlan, ChangePlanResult


class ChangePlanHandler:
    def __init__(self, uow: UnitOfWork, plans: PlanCatalog, clock: Clock) -> None:
        self._uow, self._plans, self._clock = uow, plans, clock

    def __call__(self, cmd: ChangePlan) -> ChangePlanResult:
        with self._uow as uow:
            subscription = uow.subscriptions.get(SubscriptionId(cmd.subscription_id))
            new_plan = self._plans.get(PlanCode(cmd.new_plan_code))

            delta = subscription.change_plan(new_plan, on=self._clock.today())

            uow.subscriptions.save(subscription)
            uow.commit()          # состояние + события в outbox — одной транзакцией
            # дата следующего списания = день после конца оплаченного периода:
            # смена плана её не сдвигает — это правило из раздела 1
            return ChangePlanResult(
                charged_now=delta,
                next_billing_date=subscription.period.end + timedelta(days=1),
            )

# infrastructure/unit_of_work.py
class PgUnitOfWork:
    def __init__(self, conn_factory, event_serializer) -> None:
        self._conn_factory = conn_factory
        self._serialize = event_serializer
        self._tracked: list[Subscription] = []

    def __enter__(self) -> "PgUnitOfWork":
        self._conn = self._conn_factory()
        self._conn.autocommit = False
        self.subscriptions = PgSubscriptionRepo(self._conn, track=self._tracked.append)
        return self

    def commit(self) -> None:
        # события собираются ПОСЛЕ всех изменений, но ДО COMMIT
        with self._conn.cursor() as cur:
            for aggregate in self._tracked:
                for event in aggregate.pull_events():
                    cur.execute(
                        "INSERT INTO outbox (id, aggregate_id, type, payload, occurred_at) "
                        "VALUES (%s, %s, %s, %s, %s)",
                        self._serialize(event),
                    )
        self._conn.commit()

    def __exit__(self, exc_type, *_) -> None:
        if exc_type is not None:
            self._conn.rollback()   # исключение домена = откат, без вариантов
        self._conn.close()

# infrastructure/pg_subscription_repo.py
class ConcurrencyConflict(RuntimeError): ...
class SubscriptionNotFound(LookupError): ...


class PgSubscriptionRepo:
    def __init__(self, conn, track) -> None:
        self._conn, self._track = conn, track

    def get(self, subscription_id: SubscriptionId) -> Subscription:
        with self._conn.cursor(row_factory=dict_row) as cur:
            cur.execute("SELECT * FROM subscription WHERE id = %s", (subscription_id,))
            row = cur.fetchone()
        if row is None:
            raise SubscriptionNotFound(subscription_id)

        # ручная сборка: агрегат не наследуется от базового класса ORM и ничего о ней не знает
        aggregate = Subscription(
            subscription_id=subscription_id,
            customer_id=CustomerId(row["customer_id"]),
            plan_code=PlanCode(row["plan_code"]),
            price=Money(row["price_amount"], row["price_currency"]),
            period=BillingPeriod(row["period_start"], row["period_end"]),
            status=Status(row["status"]),
            seats=row["seats"], max_seats=row["max_seats"],
            assigned_seats=row["assigned_seats"],
            trial_used=row["trial_used"], grace_until=row["grace_until"],
            version=row["version"],
        )
        self._track(aggregate)   # UoW будет знать, у кого забирать события при commit
        return aggregate

    def save(self, s: Subscription) -> None:
        with self._conn.cursor() as cur:
            cur.execute(
                "UPDATE subscription SET plan_code=%s, price_amount=%s, price_currency=%s, "
                "seats=%s, assigned_seats=%s, max_seats=%s, status=%s, period_start=%s, "
                "period_end=%s, grace_until=%s, version = version + 1 "
                "WHERE id = %s AND version = %s",
                (s.plan_code, s.price.amount, s.price.currency, s.seats, s.assigned_seats,
                 s.max_seats, s.status.value, s.period.start, s.period.end,
                 s.grace_until, s.id, s.version),
            )
            if cur.rowcount == 0:
                # либо строки нет, либо кто-то обновил её между нашими SELECT и UPDATE
                raise ConcurrencyConflict(f"версия {s.version} устарела для {s.id}")

# api/dto.py — форма запроса и ответа; про бизнес-правила здесь не знают
from decimal import Decimal
from uuid import UUID

from fastapi import APIRouter, Depends, Header, HTTPException
from pydantic import BaseModel, Field


class ChangePlanRequest(BaseModel):
    plan_code: str = Field(min_length=2, max_length=32, pattern=r"^[a-z0-9_]+$")


class ChangePlanResponse(BaseModel):
    charged_now: Decimal
    currency: str
    next_billing_date: str


# api/error_mapping.py — один словарь вместо десятка try/except в каждом эндпоинте
HTTP_STATUS: dict[type[Exception], int] = {
    SubscriptionNotFound: 404,
    InvalidTransition: 409,      # состояние не позволяет операцию
    SamePlan: 409,
    TrialAlreadyUsed: 409,
    SeatLimitExceeded: 422,      # запрос синтаксически верен, но нарушает правило
    ConcurrencyConflict: 409,    # клиенту стоит повторить запрос
}


# api/routes.py
router = APIRouter(prefix="/subscriptions", tags=["subscriptions"])


@router.post("/{subscription_id}/change-plan", response_model=ChangePlanResponse)
def change_plan(
    subscription_id: UUID,
    body: ChangePlanRequest,
    idempotency_key: str = Header(alias="Idempotency-Key"),
    handler: ChangePlanHandler = Depends(get_change_plan_handler),
) -> ChangePlanResponse:
    try:
        result = handler(ChangePlan(subscription_id, body.plan_code, idempotency_key))
    except tuple(HTTP_STATUS) as exc:
        # текст доменного исключения написан на языке предметной области —
        # его не стыдно показать клиенту
        raise HTTPException(status_code=HTTP_STATUS[type(exc)], detail=str(exc)) from exc

    # наружу уходит DTO, а не агрегат: внутренняя модель не становится публичным контрактом
    return ChangePlanResponse(
        charged_now=result.charged_now.amount,
        currency=result.charged_now.currency,
        next_billing_date=result.next_billing_date.isoformat(),
    )

# tests/unit/test_change_plan.py
import pytest
from datetime import date

NOV = BillingPeriod(date(2026, 11, 1), date(2026, 11, 30))


@pytest.mark.parametrize("day, expected", [
    (date(2026, 11, 1),  "4350.00"),   # 30/30 дней: 7500 − 3000
    (date(2026, 11, 12), "2850.00"),   # 19/30: 4750 − 1900
    (date(2026, 11, 30), "150.00"),    # 1/30:  250 − 100
])
def test_proration_is_linear_in_remaining_days(day, expected):
    sub = active_subscription(price=Money.rub("3000.00"), period=NOV)
    delta = sub.change_plan(plan("business", Money.rub("7500.00")), on=day)
    assert delta == Money.rub(expected)


def test_downgrade_below_assigned_seats_is_rejected():
    sub = active_subscription(assigned_seats=12)
    with pytest.raises(SeatLimitExceeded, match="освободите места"):
        sub.change_plan(plan("starter", Money.rub("900.00"), max_seats=5),
                        on=date(2026, 11, 12))


def test_expired_subscription_never_changes_plan():
    sub = expired_subscription()
    with pytest.raises(InvalidTransition):
        sub.change_plan(plan("business", Money.rub("7500.00")), on=date(2026, 11, 12))

class FakeUnitOfWork:
    """20 строк вместо контейнера с Postgres — и тест сценария бежит за миллисекунду."""

    def __init__(self, *subscriptions):
        self.subscriptions, self.committed = FakeRepo(subscriptions), False

    def __enter__(self): return self
    def __exit__(self, *exc): pass
    def commit(self): self.committed = True


def test_handler_does_not_commit_when_domain_rejects():
    uow = FakeUnitOfWork(expired_subscription(id=SUB_ID))
    handler = ChangePlanHandler(uow, FakePlans(), FrozenClock(date(2026, 11, 12)))

    with pytest.raises(InvalidTransition):
        handler(ChangePlan(SUB_ID, "business", "idem-1"))

    assert not uow.committed          # откат обязателен

def test_concurrent_updates_conflict(pg_uow_factory):
    with pg_uow_factory() as uow_a, pg_uow_factory() as uow_b:
        a = uow_a.subscriptions.get(SUB_ID)
        b = uow_b.subscriptions.get(SUB_ID)     # обе транзакции видят version = 7
        a.assign_seats(5); uow_a.subscriptions.save(a); uow_a.commit()
        b.assign_seats(9)
        with pytest.raises(ConcurrencyConflict):
            uow_b.subscriptions.save(b)