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.

class UserSettingsUpdatedEvent(BaseEvent):
    model_config = ConfigDict(extra="allow")

    event_type: Literal[EventType.USER_SETTINGS_UPDATED] = EventType.USER_SETTINGS_UPDATED
    user_id: str
    changed_fields: list[str] = Field(default_factory=list)
    reason: str | None = None

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:

_settings_fields = set(DomainUserSettings.__dataclass_fields__)

Updating settings¶

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

    async def update_user_settings(
        self, user_id: str, updates: DomainUserSettingsUpdate, reason: str | None = None
    ) -> DomainUserSettings:
        """Upsert provided fields into current settings, publish minimal event, and cache."""
        current = await self.get_user_settings(user_id)

        changes = {k: v for k, v in dataclasses.asdict(updates).items() if v is not None}
        if not changes:
            return current

        new_settings = self._build_settings({
            **dataclasses.asdict(current),
            **changes,
            "version": (current.version or 0) + 1,
            "updated_at": datetime.now(timezone.utc),
        })

        await self._publish_settings_event(user_id, changes, reason)

        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

Applying events¶

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

    def _apply_event(self, settings: DomainUserSettings, event: DomainUserSettingsChangedEvent) -> DomainUserSettings:
        """Apply a settings update event via dict merge."""
        event_data = {k: v for k, v in dataclasses.asdict(event).items() if v is not None}
        return self._build_settings({**dataclasses.asdict(settings), **event_data, "updated_at": event.timestamp})

The validate_python call handles nested dict-to-dataclass conversion, enum parsing, and type coercion automatically. See Domain 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: TTLCache[str, DomainUserSettings] = TTLCache(
            maxsize=self._max_cache_size,
            ttl=self._cache_ttl.total_seconds(),
        )

Cache invalidation happens through TTL-based expiration. The cache has a configurable TTL, so stale entries are automatically refreshed:

    async def get_user_settings_fresh(self, user_id: str) -> DomainUserSettings:
        """Bypass cache and rebuild settings from snapshot + events."""
        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)
        return settings

After each update, the service updates its local cache directly. Other instances pick up changes when their cache TTL expires.

Settings history¶

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

    async def get_settings_history(self, user_id: str, limit: int = 50) -> list[DomainSettingsHistoryEntry]:
        """Get history from changed fields recorded in events."""
        events = await self.repository.get_settings_events(user_id, [EventType.USER_SETTINGS_UPDATED], limit=limit)
        history: list[DomainSettingsHistoryEntry] = []
        for event in events:
            for fld in event.changed_fields:
                history.append(
                    DomainSettingsHistoryEntry(
                        timestamp=event.timestamp,
                        event_type=event.event_type,
                        field=f"/{fld}",
                        old_value=None,
                        new_value=getattr(event, fld, None),
                        reason=event.reason,
                    )
                )
        return history

Key files¶

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

Domain Dataclasses — domain model conventions and dict-to-model conversion