Skip to content

User settings events

This document explains how user settings are stored, updated, and reconstructed using event sourcing with a unified event type and TypeAdapter-based merging.

Unified event approach

All user settings changes emit a single USER_SETTINGS_UPDATED event type. There are no specialized events for theme, notifications, or editor settings. This eliminates branching in both publishing and consuming code.

The changed_fields list identifies which settings changed. Typed fields (theme, notifications, editor, etc.) contain the new values as Pydantic model fields.

TypeAdapter pattern

The service uses Pydantic's TypeAdapter for dict-based operations without reflection or branching:

    def __init__(
        self,

Updating settings

The update_user_settings method merges changes into current settings, publishes an event, and manages snapshots:

        self._add_to_cache(user_id, new_settings)
        if (await self.repository.count_events_since_snapshot(user_id)) >= 10:
            await self.repository.create_snapshot(new_settings)
        return new_settings

    async def _publish_settings_event(self, user_id: str, changes: dict[str, Any], reason: str | None) -> None:
        """Publish settings update event with typed payload fields."""
        await self.event_service.publish_event(
            event_type=EventType.USER_SETTINGS_UPDATED,
            aggregate_id=f"user_settings_{user_id}",
            payload={
                "user_id": user_id,
                "changed_fields": list(changes.keys()),
                "reason": reason,
                **changes,
            },
            metadata=None,
        )

    async def update_theme(self, user_id: str, theme: Theme) -> DomainUserSettings:
        """Update user's theme preference"""
        return await self.update_user_settings(
            user_id, DomainUserSettingsUpdate(theme=theme), reason="User changed theme"
        )

    async def update_notification_settings(
        self, user_id: str, notifications: DomainNotificationSettings

Applying events

When reconstructing settings from events, _apply_event merges each event's changes:

The validate_python call handles nested dict-to-dataclass conversion, enum parsing, and type coercion automatically. See Pydantic Dataclasses for details.

Settings reconstruction

graph TB
    A[get_user_settings] --> B{Cache hit?}
    B -->|Yes| C[Return cached]
    B -->|No| D[Load snapshot]
    D --> E[Query events since snapshot]
    E --> F[Apply each event]
    F --> G[Cache result]
    G --> C

Snapshots are created automatically when event count exceeds threshold (10 events). This bounds reconstruction cost while preserving full event history for auditing.

Cache layer

Settings are cached with TTL to avoid repeated reconstruction:

        self._cache_ttl = timedelta(minutes=5)
        self._max_cache_size = 1000
        self._cache: TTLCache[str, DomainUserSettings] = TTLCache(
            maxsize=self._max_cache_size,
            ttl=self._cache_ttl.total_seconds(),
        )

        self.logger.info(

Cache invalidation happens via EventBus subscription. The EventBus filters out self-published messages, so the handler only runs for events from other instances:

        snapshot = await self.repository.get_snapshot(user_id)

        settings: DomainUserSettings
        event_types = [EventType.USER_SETTINGS_UPDATED]
        if snapshot:
            settings = snapshot
            events = await self.repository.get_settings_events(user_id, event_types, since=snapshot.updated_at)
        else:
            settings = DomainUserSettings(user_id=user_id)
            events = await self.repository.get_settings_events(user_id, event_types)

        for event in events:
            settings = self._apply_event(settings, event)

        self._add_to_cache(user_id, settings)

After each update, the service updates its local cache directly, then publishes to the event bus to trigger cache invalidation on other instances.

Settings history

The get_settings_history method returns a list of changes extracted from events:

        settings = DomainUserSettings(user_id=user_id)
        for event in events:
            settings = self._apply_event(settings, event)

        await self.repository.create_snapshot(settings)
        self._add_to_cache(user_id, settings)

        await self.event_service.publish_event(
            event_type=EventType.USER_SETTINGS_UPDATED,
            aggregate_id=f"user_settings_{user_id}",
            payload={
                "user_id": user_id,
                "changed_fields": [],
                "reason": f"Settings restored to {timestamp.isoformat()}",
            },
            metadata=None,
        )

Key files

File Purpose
services/user_settings_service.py Settings service with caching and event sourcing
services/event_bus.py Cross-instance event distribution
domain/user/settings_models.py DomainUserSettings, DomainUserSettingsUpdate dataclasses
domain/events/typed.py UserSettingsUpdatedEvent definition
db/repositories/user_settings_repository.py Snapshot and event queries