Event Sourcing e CQRS: quando eventos são a fonte da verdade

# bank_account.py — agregado em Event Sourcing

from dataclasses import dataclass, field
from typing import Literal

# ─── Events (imutáveis) ───
@dataclass(frozen=True)
class AccountOpened:
    account_id: str
    owner: str

@dataclass(frozen=True)
class Credited:
    account_id: str
    amount: float

@dataclass(frozen=True)
class Debited:
    account_id: str
    amount: float

Event = AccountOpened | Credited | Debited


# ─── Aggregate ───
@dataclass
class BankAccount:
    account_id: str = ""
    owner: str = ""
    balance: float = 0.0
    version: int = 0                    # número do último evento aplicado
    _uncommitted: list[Event] = field(default_factory=list)

    @classmethod
    def load(cls, history: list[Event]) -> "BankAccount":
        acc = cls()
        for e in history:
            acc._apply(e)
            acc.version += 1
        return acc

    def _apply(self, e: Event) -> None:
        if isinstance(e, AccountOpened):
            self.account_id = e.account_id
            self.owner = e.owner
        elif isinstance(e, Credited):
            self.balance += e.amount
        elif isinstance(e, Debited):
            self.balance -= e.amount

    # ─── Commands (validam + produzem eventos) ───
    def open(self, account_id: str, owner: str) -> None:
        if self.account_id:
            raise ValueError("Already opened")
        self._raise(AccountOpened(account_id, owner))

    def credit(self, amount: float) -> None:
        if amount <= 0:
            raise ValueError("Amount must be positive")
        self._raise(Credited(self.account_id, amount))

    def debit(self, amount: float) -> None:
        if amount <= 0:
            raise ValueError("Amount must be positive")
        if self.balance - amount < 0:
            raise ValueError("Insufficient funds")   # regra de negócio
        self._raise(Debited(self.account_id, amount))

    def _raise(self, e: Event) -> None:
        self._apply(e)
        self._uncommitted.append(e)
        self.version += 1


# ─── Repository ───
class AccountRepository:
    def __init__(self, event_store):
        self.store = event_store

    async def load(self, account_id: str) -> BankAccount:
        events = await self.store.load_stream(account_id)
        return BankAccount.load(events)

    async def save(self, acc: BankAccount) -> None:
        if not acc._uncommitted:
            return
        expected_version = acc.version - len(acc._uncommitted)
        await self.store.append(
            acc.account_id, acc._uncommitted, expected_version
        )
        acc._uncommitted.clear()

# projections.py — projector async alimentando tabela de leitura

async def account_balance_projector(event_stream):
    """Projeção simples: saldo atual por conta."""
    async for event in event_stream:
        async with db.begin() as tx:
            if isinstance(event, AccountOpened):
                await tx.execute(
                    text("INSERT INTO accounts_view (id, owner, balance) VALUES (:id, :o, 0)"),
                    {"id": event.account_id, "o": event.owner},
                )
            elif isinstance(event, Credited):
                await tx.execute(
                    text("UPDATE accounts_view SET balance = balance + :a WHERE id = :id"),
                    {"id": event.account_id, "a": event.amount},
                )
            elif isinstance(event, Debited):
                await tx.execute(
                    text("UPDATE accounts_view SET balance = balance - :a WHERE id = :id"),
                    {"id": event.account_id, "a": event.amount},
                )
            # ledger de checkpoint: última offset processada (pra replay/resume)
            await tx.execute(
                text("UPDATE projection_offsets SET last_event_id = :id WHERE name = 'balance'"),
                {"id": event.id},
            )


async def transactions_history_projector(event_stream):
    """Outra projeção do MESMO stream: histórico de operações."""
    async for event in event_stream:
        if isinstance(event, (Credited, Debited)):
            await db.execute(
                text("""INSERT INTO transactions_view
                        (account_id, type, amount, at)
                        VALUES (:id, :t, :a, :ts)"""),
                {
                    "id": event.account_id,
                    "t": type(event).__name__,
                    "a": getattr(event, "amount", 0),
                    "ts": event.timestamp,
                },
            )

# snapshot.py — carregamento com snapshot

async def load_with_snapshot(account_id: str) -> BankAccount:
    # 1. Busca snapshot mais recente
    snapshot = await db.fetchrow(
        "SELECT state, at_version FROM snapshots WHERE aggregate_id = $1 ORDER BY at_version DESC LIMIT 1",
        account_id,
    )

    if snapshot:
        acc = BankAccount(**snapshot["state"])
        acc.version = snapshot["at_version"]
        start_version = snapshot["at_version"] + 1
    else:
        acc = BankAccount()
        start_version = 0

    # 2. Carrega eventos APÓS o snapshot
    events = await load_events(account_id, from_version=start_version)
    for e in events:
        acc._apply(e)
        acc.version += 1

    return acc


async def save_snapshot_if_needed(acc: BankAccount):
    # A cada 100 eventos, persiste snapshot
    if acc.version % 100 == 0:
        await db.execute(
            "INSERT INTO snapshots (aggregate_id, state, at_version) VALUES ($1, $2, $3)",
            acc.account_id, serialize(acc), acc.version,
        )

# upcaster.py — exemplo de upcaster
def upcast(raw: dict) -> Event:
    event_type = raw["type"]
    version = raw.get("version", 1)

    if event_type == "Credited":
        if version == 1:
            # v1 não tinha currency. Antes, todos BRL.
            raw["currency"] = "BRL"
            raw["version"] = 2
        return Credited(**raw)

    if event_type == "Debited":
        ...

-- event store simples em Postgres
CREATE TABLE events (
    id                BIGSERIAL PRIMARY KEY,            -- ordem global
    aggregate_id      UUID NOT NULL,
    aggregate_version INT NOT NULL,                     -- ordem dentro do aggregate
    event_type        TEXT NOT NULL,
    payload           JSONB NOT NULL,
    metadata          JSONB DEFAULT '{}',               -- correlação, causação, user
    occurred_at       TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    UNIQUE(aggregate_id, aggregate_version)             -- optimistic concurrency
);

CREATE INDEX ON events (aggregate_id, aggregate_version);
CREATE INDEX ON events (event_type);                    -- por tipo p/ projeções
CREATE INDEX ON events (occurred_at);                   -- filtros temporais

-- append com otimista
INSERT INTO events (aggregate_id, aggregate_version, event_type, payload)
VALUES ($1, $2, $3, $4);
-- vai falhar com "unique constraint violation" se outro writer já gravou aquela versão