mantimon-tcg/backend/app/services

Services Layer

Business logic layer between API endpoints and data access.

Architecture Principles

• Stateless: Config comes from request, not server state
• DI Pattern: Dependencies injected via constructor
• Repository Protocol: Data access through abstract protocols
• Offline Fork Ready: No hard dependencies on Postgres/Redis

Services Overview

Service	Responsibility	Dependencies
CardService	Card definition lookup (singleton)	JSON files
CollectionService	User card ownership CRUD	CollectionRepository, CardService
DeckService	Deck CRUD + validation	DeckRepository, CollectionRepository, CardService
DeckValidator	Pure deck validation functions	CardService (for lookups)
UserService	User profile management	Direct DB access
GameStateManager	Game state (Redis + Postgres)	Redis, AsyncSession
GameService	Game orchestration + actions	GameStateManager, CardService, engine_factory
ConnectionManager	WebSocket connection tracking	redis_factory
JWTService	Token creation/verification	Settings
TokenStore	Refresh token storage	Redis

Key Patterns

Config from Request

The backend is stateless. Rules come from the request via config objects like DeckConfig:

# Correct - config from caller (frontend provides rules)
await deck_service.create_deck(
    user_id=user.id,
    cards=request.cards,
    deck_config=request.deck_config,  # From request body
    max_decks=user.max_decks,
)

# Wrong - baked-in config
class DeckService:
    def __init__(self, config: DeckConfig):  # Don't do this
        self._config = config

This enables different game modes (campaign, freeplay, custom) without server-side config.

UNSET Sentinel

For nullable fields that can be explicitly cleared, use UNSET to distinguish between "not provided" and "set to null":

from app.repositories.protocols import UNSET

async def update_deck(
    self,
    deck_id: UUID,
    name: str | None = None,           # None = don't change
    description: str | None = UNSET,   # UNSET = keep, None = clear, str = set
) -> DeckEntry:
    ...
    if description is not UNSET:
        # User explicitly provided a value (could be None to clear)
        record.description = description

API layer usage with Pydantic:

# Check if field was in the request payload
description = request.description if "description" in request.model_fields_set else UNSET

Repository Injection

Services use constructor injection with repository protocols:

class DeckService:
    def __init__(
        self,
        deck_repository: DeckRepository,      # Protocol, not implementation
        card_service: CardService,
        collection_repository: CollectionRepository | None = None,
    ) -> None:
        self._deck_repo = deck_repository
        self._card_service = card_service
        self._collection_repo = collection_repository

Benefits:

• Testability: Inject mock repositories
• Offline Fork: Swap PostgresRepository for LocalRepository
• Decoupling: Services don't know about SQLAlchemy

Factory Injection for External Resources

For external resources (Redis, GameEngine), inject factory functions rather than instances:

from collections.abc import AsyncIterator, Callable

# Type aliases for factories
RedisFactory = Callable[[], AsyncIterator["Redis"]]
EngineFactory = Callable[[GameState], GameEngine]

class ConnectionManager:
    """External resources injected as factories."""

    def __init__(
        self,
        conn_ttl_seconds: int = 3600,
        redis_factory: RedisFactory | None = None,
    ) -> None:
        self.conn_ttl_seconds = conn_ttl_seconds
        self._get_redis = redis_factory if redis_factory is not None else get_redis

class GameService:
    """Engine created via injected factory for testability."""

    def __init__(
        self,
        state_manager: GameStateManager | None = None,
        engine_factory: EngineFactory | None = None,
    ) -> None:
        self._state_manager = state_manager or game_state_manager
        self._engine_factory = engine_factory or self._default_engine_factory

Global singletons are created for production, but constructors always support injection:

# Production singleton
connection_manager = ConnectionManager()

# Tests inject mocks - no patching needed
manager = ConnectionManager(redis_factory=mock_redis_factory)

Pure Validation Functions

Validation logic is extracted into pure functions for reuse:

# app/services/deck_validator.py
def validate_deck(
    cards: dict[str, int],
    energy_cards: dict[str, int],
    deck_config: DeckConfig,
    card_lookup: Callable[[str], CardDefinition | None],
    owned_cards: dict[str, int] | None = None,
) -> ValidationResult:
    """Pure function - all inputs from caller."""
    ...

Service Details

CardService

Singleton that loads card definitions from JSON files at startup.

from app.services.card_service import get_card_service

card_service = get_card_service()
card = card_service.get_card("a1-001-bulbasaur")
all_cards = card_service.get_all_cards()

CollectionService

Manages user card ownership with source tracking.

# Add cards to collection
await collection_service.add_cards(
    user_id=user.id,
    card_definition_id="a1-001-bulbasaur",
    quantity=4,
    source=CardSource.BOOSTER,
)

# Get owned cards for deck validation
owned = await collection_service.get_owned_cards_dict(user.id)

DeckService

Deck CRUD with validation and ownership checks.

# Create deck - validates and stores result
deck = await deck_service.create_deck(
    user_id=user.id,
    name="My Deck",
    cards={"a1-001-bulbasaur": 4, ...},
    energy_cards={"grass": 14, "colorless": 6},
    deck_config=DeckConfig(),
    max_decks=user.max_decks,
    validate_ownership=True,  # Campaign mode
)

# Deck can be saved even if invalid (for work-in-progress)
if not deck.is_valid:
    print(deck.validation_errors)

GameStateManager

Hybrid storage: Redis for fast access, Postgres for durability.

manager = GameStateManager(redis, db_session)

# Save to Redis (fast, every action)
await manager.save_to_redis(game_id, game_state)

# Persist to Postgres (at turn boundaries)
await manager.persist_to_db(game_id, game_state)

Testing

Directory	Purpose	Database
tests/unit/services/	Pure unit tests, mocked deps	No
tests/services/	Integration tests	Yes (testcontainers)

No monkey patching - Always use constructor DI for mocks:

# WRONG - monkey patching
async def test_something():
    with patch.object(service, "_method") as mock:
        ...

# CORRECT - constructor injection
@pytest.fixture
def mock_redis():
    return AsyncMock()

@pytest.fixture
def connection_manager(mock_redis):
    @asynccontextmanager
    async def mock_redis_factory():
        yield mock_redis
    return ConnectionManager(redis_factory=mock_redis_factory)

Unit Test Example

def test_validate_deck_requires_basic_pokemon():
    """Test that decks must contain at least one Basic Pokemon."""
    result = validate_deck(
        cards={"trainer-card": 40},  # No Pokemon
        energy_cards={"colorless": 20},
        deck_config=DeckConfig(),
        card_lookup=mock_lookup,
    )
    assert not result.is_valid
    assert "Basic Pokemon" in str(result.errors)

Integration Test Example

@pytest.mark.asyncio
async def test_create_deck_enforces_limit(db_session, deck_service):
    """Test that free users are limited to 5 decks."""
    user = await UserFactory.create(db_session, max_decks=5)

    # Create 5 decks
    for i in range(5):
        await deck_service.create_deck(user_id=user.id, ...)

    # 6th should fail
    with pytest.raises(DeckLimitExceededError):
        await deck_service.create_deck(user_id=user.id, ...)

See Also

• app/repositories/protocols.py - Repository interfaces and DTOs
• app/repositories/postgres/ - PostgreSQL implementations
• app/core/config.py - DeckConfig and RulesConfig
• CLAUDE.md - Architecture guidelines