Skip to main content

Aggregate Examples

Real handler examples from the poker domain. All code is from the actual examples/ directory.

⍼ Angzarr supports two implementation styles:

StyleDescriptionBest For
OO (Object-Oriented)Aggregate class with @handles/[Handles] decoratorsRich domain models, encapsulation
ImperativePure guard()/validate()/compute() functionsSimple handlers, easy testing
LanguageOOImperative
Python✓—
C#✓✓
Rust✓✓
Java✓✓
Go—✓
C++—✓

Player: Reserve Funds​

The Player aggregate demonstrates the two-phase reservation pattern. Funds are reserved before joining a table, then released if the join fails.

examples/python/player/agg/handlers/player.py

@handles(player_proto.ReserveFunds)
def reserve(self, cmd: player_proto.ReserveFunds) -> player_proto.FundsReserved:
    """Reserve funds for a table buy-in."""
    if not self.exists:
        raise CommandRejectedError("Player does not exist")

    amount = cmd.amount.amount if cmd.amount else 0
    if amount <= 0:
        raise CommandRejectedError("amount must be positive")

    table_key = cmd.table_root.hex()
    if table_key in self.table_reservations:
        raise CommandRejectedError("Funds already reserved for this table")

    if amount > self.available_balance:
        raise CommandRejectedError("Insufficient funds")

    new_reserved = self.reserved_funds + amount
    new_available = self.bankroll - new_reserved
    return player_proto.FundsReserved(
        amount=cmd.amount,
        table_root=cmd.table_root,
        new_available_balance=poker_types.Currency(
            amount=new_available, currency_code="CHIPS"
        ),
        new_reserved_balance=poker_types.Currency(
            amount=new_reserved, currency_code="CHIPS"
        ),
        reserved_at=now(),
    )

State Building​

State is rebuilt by applying events. ⍼ Angzarr provides two patterns:

PatternDescriptionBest For
OO with @appliesDecorators on aggregate class methodsRich domain models
StateRouterFluent builder with typed event handlersClean separation, easy testing

examples/python/player/agg/handlers/player.py

@applies(player_proto.PlayerRegistered)
def apply_registered(self, state: _PlayerState, event: player_proto.PlayerRegistered):
    """Apply PlayerRegistered event to state."""
    state.player_id = f"player_{event.email}"
    state.display_name = event.display_name
    state.email = event.email
    state.player_type = event.player_type
    state.ai_model_id = event.ai_model_id
    state.status = "active"
    state.bankroll = 0
    state.reserved_funds = 0

@applies(player_proto.FundsDeposited)
def apply_deposited(self, state: _PlayerState, event: player_proto.FundsDeposited):
    """Apply FundsDeposited event to state."""
    if event.new_balance:
        state.bankroll = event.new_balance.amount

@applies(player_proto.FundsWithdrawn)
def apply_withdrawn(self, state: _PlayerState, event: player_proto.FundsWithdrawn):
    """Apply FundsWithdrawn event to state."""
    if event.new_balance:
        state.bankroll = event.new_balance.amount

@applies(player_proto.FundsReserved)
def apply_reserved(self, state: _PlayerState, event: player_proto.FundsReserved):
    """Apply FundsReserved event to state."""
    if event.new_reserved_balance:
        state.reserved_funds = event.new_reserved_balance.amount
    table_key = event.table_root.hex()
    if event.amount:
        state.table_reservations[table_key] = event.amount.amount

@applies(player_proto.FundsReleased)
def apply_released(self, state: _PlayerState, event: player_proto.FundsReleased):
    """Apply FundsReleased event to state."""
    if event.new_reserved_balance:
        state.reserved_funds = event.new_reserved_balance.amount
    table_key = event.table_root.hex()
    state.table_reservations.pop(table_key, None)

@applies(player_proto.FundsTransferred)
def apply_transferred(self, state: _PlayerState, event: player_proto.FundsTransferred):
    """Apply FundsTransferred event to state."""
    if event.new_balance:
        state.bankroll = event.new_balance.amount

Testing​

⍼ Angzarr uses Gherkin feature files as living specifications. The same feature files run against all language implementations, guaranteeing identical business behavior.

examples/features/unit/player.feature

Feature: Player aggregate logic
  The Player aggregate manages a player's bankroll and table reservations.
  It's the source of truth for how much money a player has and where it's allocated.
Scenario: Register a new human player
  Given no prior events for the player aggregate
  When I handle a RegisterPlayer command with name "Alice" and email "alice@example.com"
  Then the result is a PlayerRegistered event
  And the player event has display_name "Alice"
  And the player event has player_type "HUMAN"

Scenario: Cannot register player twice
  Given a PlayerRegistered event for "Alice"
  When I handle a RegisterPlayer command with name "Alice2" and email "alice@example.com"
  Then the command fails with status "FAILED_PRECONDITION"
  And the error message contains "already exists"
Scenario: Reserve funds for table buy-in
  Given a PlayerRegistered event for "Alice"
  And a FundsDeposited event with amount 1000
  When I handle a ReserveFunds command with amount 500 for table "table-1"
  Then the result is a FundsReserved event
  And the player event has amount 500
  And the player event has new_available_balance 500

examples/python/tests/unit/test_player.py

import sys
from pathlib import Path

import pytest
from pytest_bdd import scenarios, given, when, then, parsers
from google.protobuf.any_pb2 import Any as ProtoAny
# Load scenarios from feature file
scenarios("../../../features/unit/player.feature")
@given(parsers.parse('a PlayerRegistered event for "{name}"'))
def player_registered_event(ctx, name):
    """Add a PlayerRegistered event."""
    event = player.PlayerRegistered(
        display_name=name,
        email=f"{name.lower()}@example.com",
        player_type=poker_types.HUMAN,
        registered_at=make_timestamp(),
    )
    ctx.add_event(event)
@when(parsers.parse('I handle a RegisterPlayer command with name "{name}" and email "{email}"'))
def handle_register_player_cmd(ctx, name, email):
    """Handle RegisterPlayer command."""
    cmd = player.RegisterPlayer(
        display_name=name,
        email=email,
        player_type=poker_types.HUMAN,
    )
    _handle_command(ctx, cmd, handle_register_player)
@then("the result is a PlayerRegistered event")
def result_is_player_registered(ctx):
    """Verify result is PlayerRegistered event."""
    assert ctx.result is not None, f"Expected result, got error: {ctx.error}"
    assert len(ctx.result.pages) == 1
    event_any = ctx.result.pages[0].event
    assert event_any.type_url.endswith("PlayerRegistered")

Run tests:

# Python - pytest-bdd
cd examples/python && pytest tests/unit/test_player.py

# Rust - cucumber-rs
cd examples/rust/tests && cargo test --test player

# Go - godog
cd examples/go && go test -v ./... --godog.tags=@player

Next Steps​