Source code for aeat.application.user_profile._lifecycle

"""Canonical lifecycle service for the schema-driven user-profile backend.

This service is the single sanctioned write path for live user-profile
values. It routes every register / edit / remove / duplicate / list /
read operation through the secure-DB repository and the schema-aware
validation service. CLI thin adapters and downstream consumers call
this surface; no caller should construct :class:`UserProfileRecord`
aggregates or touch the secure repository directly. Every mutating
operation emits a typed bucket event to :class:`BucketEventHistoryRepository`.
"""

from __future__ import annotations

from collections.abc import Iterable
from datetime import date, datetime
from typing import Literal

from ...adapters.persistence.profile.buckets import BucketEventHistoryRepository
from ...core.errors import BaseSeverity
from ...core.hashing import sha256_hex
from ...domain.buckets import (
    BucketEvent,
    BucketEventHistoryRepositoryProtocol,
    BucketEventObjectType,
    BucketEventType,
    append_bucket_event,
    derive_bucket_event_id,
)
from ...domain.user_profile import (
    ProfileAlreadyExistsError,
    ProfileNotFoundError,
    ProfileSchemaValidationError,
    UserProfileFact,
    UserProfileRecord,
    UserProfileStatus,
    utc_now,
)
from . import (
    DuplicateProfileCommand,
    EditProfileFieldCommand,
    ProfileLifecycleResult,
    ProfileListing,
    ProfileListResult,
    ReactivateProfileCommand,
    RegisterProfileCommand,
    RemoveProfileCommand,
    RenameProfileCommand,
)
from ._repository import UserProfileLifecycleRepository
from ._validation import ProfileValidationService

_PROFILE_LIFECYCLE_ACTOR = "aeat.application.user_profile"
_PROFILE_ALREADY_EXISTS_MESSAGE = "profile already exists in the active bucket"
_PROFILE_TOMBSTONED_RENAME_MESSAGE = "tombstoned profile cannot be renamed"
_PROFILE_TOMBSTONED_DUPLICATE_MESSAGE = "tombstoned profile cannot be duplicated"
_PROFILE_NOT_TOMBSTONED_MESSAGE = "profile is not tombstoned; reactivate refuses a live profile"
_PROFILE_SCHEMA_VALIDATION_MESSAGE = "profile facts failed schema validation"


def _paths_payload(facts: tuple[UserProfileFact, ...]) -> dict[str, str]:
    """Return bounded path summary payload for a bucket event."""
    paths = sorted({fact.path for fact in facts})
    encoded = "\n".join(paths).encode("utf-8")
    return {"path_count": str(len(paths)), "paths_sha256": sha256_hex(encoded)}


[docs] class ProfileLifecycleService: """Schema-validated, secure-DB-backed user-profile lifecycle service.""" def __init__( self, *, repository: UserProfileLifecycleRepository, validator: ProfileValidationService, events: BucketEventHistoryRepositoryProtocol | None = None, ) -> None: self._repository = repository self._validator = validator self._events = events or BucketEventHistoryRepository() # ── register / list / read ─────────────────────────────────────
[docs] def register(self, command: RegisterProfileCommand) -> ProfileLifecycleResult: """Register a new active profile aggregate. Returns a :class:`ProfileLifecycleResult`. """ if self._repository.exists(command.profile_id): raise _profile_already_exists_error( profile_id=command.profile_id, bucket_id=self._repository.bucket_id, ) self._reject_invalid(command.profile_id, command.facts) now = utc_now() record = UserProfileRecord( schema_id=self._validator.schema.id, schema_version=self._validator.schema.version, profile_id=command.profile_id, display_name=command.display_name, facts=command.facts, created_at=now, updated_at=now, ) self._repository.save(record) self._emit_event( event_type=BucketEventType.PROFILE_BUCKET_CREATED, object_id=record.profile_id, occurred_at=now, payload={"display_name": record.display_name}, ) if record.facts: self._emit_event( event_type=BucketEventType.PROFILE_VALUES_UPDATED, object_id=record.profile_id, occurred_at=now, payload=_paths_payload(record.facts), ) return ProfileLifecycleResult(profile=record, applied_at=now)
[docs] def read(self, profile_id: str) -> UserProfileRecord: """Return the live :class:`UserProfileRecord` aggregate or raise :class:`ProfileNotFoundError`.""" return self._repository.load(profile_id)
[docs] def list_profiles(self) -> ProfileListResult: """List every profile currently visible in the active bucket. Returns a :class:`ProfileListResult` with all non-tombstoned profiles. """ listings: list[ProfileListing] = [] for record in self._iter_profiles(): listings.append( ProfileListing( profile_id=record.profile_id, display_name=record.display_name, status=record.status, created_at=record.created_at, updated_at=record.updated_at, ), ) listings.sort(key=lambda row: row.profile_id) return ProfileListResult(profiles=tuple(listings))
# ── edit / remove / duplicate ──────────────────────────────────
[docs] def edit_field(self, command: EditProfileFieldCommand) -> ProfileLifecycleResult: """Upsert one effective-dated fact into a profile aggregate and return a :class:`ProfileLifecycleResult`.""" record = self._repository.load(command.profile_id) new_fact = UserProfileFact( path=command.path, value=command.value, valid_from=command.valid_from, valid_to=command.valid_to, source=command.source, ) next_facts = self._merge_facts(record.facts, (new_fact,)) self._reject_invalid(command.profile_id, next_facts) result = self._save_updated(record, next_facts) event_type = ( BucketEventType.PROFILE_VALUES_CLEARED if new_fact.value is None else BucketEventType.PROFILE_VALUES_UPDATED ) self._emit_event( event_type=event_type, object_id=record.profile_id, occurred_at=result.applied_at, payload={"path": new_fact.path}, ) return result
[docs] def remove(self, command: RemoveProfileCommand) -> ProfileLifecycleResult: """Tombstone the live root. Immutable filing snapshots are retained. Returns a :class:`ProfileLifecycleResult` with the tombstoned profile. """ record = self._repository.load(command.profile_id) tombstoned = record.tombstone() self._repository.save(tombstoned) self._emit_event( event_type=BucketEventType.PROFILE_TOMBSTONED, object_id=tombstoned.profile_id, occurred_at=tombstoned.updated_at, ) return ProfileLifecycleResult(profile=tombstoned, applied_at=tombstoned.updated_at)
[docs] def reactivate(self, command: ReactivateProfileCommand) -> ProfileLifecycleResult: """Restore a tombstoned root to active status; symmetric inverse of :meth:`remove`. Refuses if the target is not currently tombstoned, so a reactivate can never silently no-op against an already-active profile. Returns a :class:`ProfileLifecycleResult` with the reactivated profile. """ record = self._repository.load(command.profile_id) if record.status is not UserProfileStatus.TOMBSTONED: raise _profile_not_tombstoned_error(command.profile_id) reactivated = record.reactivate() self._repository.save(reactivated) self._emit_event( event_type=BucketEventType.PROFILE_REACTIVATED, object_id=reactivated.profile_id, occurred_at=reactivated.updated_at, ) return ProfileLifecycleResult(profile=reactivated, applied_at=reactivated.updated_at)
[docs] def rename(self, command: RenameProfileCommand) -> ProfileLifecycleResult: """Update a live profile's display label and return a :class:`ProfileLifecycleResult`. Profile identity is an immutable UUID, so a rename touches only the operator-visible ``display_name``: the record is loaded, its label updated, and re-saved under the same secure-object key. There is no re-key, no directory move, and no rollback machinery. Refuses if the profile is tombstoned. Emits ``PROFILE_RENAMED`` carrying the prior label so the audit trail records the relabel. The orchestration layer updates the parallel copy of the label held in the plaintext bucket manifest; the service contract is record-only. """ source = self._repository.load(command.profile_id) if source.status is not UserProfileStatus.ACTIVE: raise _profile_tombstoned_error(command.profile_id, action="rename") if command.target_display_name == source.display_name: return ProfileLifecycleResult(profile=source, applied_at=utc_now()) now = utc_now() target = source.model_copy( update={ "display_name": command.target_display_name, "updated_at": now, }, ) self._repository.save(target) self._emit_event( event_type=BucketEventType.PROFILE_RENAMED, object_id=target.profile_id, occurred_at=now, payload={"previous_display_name": source.display_name}, ) return ProfileLifecycleResult(profile=target, applied_at=now)
[docs] def duplicate(self, command: DuplicateProfileCommand) -> ProfileLifecycleResult: """Copy an existing live profile under a new id and display name. Returns a :class:`ProfileLifecycleResult` with the newly created duplicate profile. """ if self._repository.exists(command.target_profile_id): raise _profile_already_exists_error( profile_id=command.target_profile_id, bucket_id=self._repository.bucket_id, ) source = self._repository.load(command.source_profile_id) if source.status is not UserProfileStatus.ACTIVE: raise _profile_tombstoned_error(command.source_profile_id, action="duplicate") now = utc_now() target = source.model_copy( update={ "profile_id": command.target_profile_id, "display_name": command.target_display_name, "created_at": now, "updated_at": now, "removed_at": None, "status": UserProfileStatus.ACTIVE, }, ) self._repository.save(target) self._emit_event( event_type=BucketEventType.PROFILE_DUPLICATED, object_id=target.profile_id, occurred_at=now, payload={"source_profile_id": source.profile_id}, ) return ProfileLifecycleResult(profile=target, applied_at=now)
# ── helpers ──────────────────────────────────────────────────── def _reject_invalid(self, profile_id: str, facts: Iterable[UserProfileFact]) -> None: report = self._validator.validate_facts(profile_id, facts) blocking = [issue for issue in report.issues if issue.severity is BaseSeverity.ERROR] if blocking: raise ProfileSchemaValidationError( _PROFILE_SCHEMA_VALIDATION_MESSAGE, context={ "profile_id": profile_id, "issue_count": len(blocking), "issue_codes": tuple(issue.code for issue in blocking), "issue_paths": tuple(issue.path for issue in blocking if issue.path is not None), "schema_version": report.schema_version, }, translated_message="application.user_profile.errors.lifecycle_schema_validation_failed", ) def _save_updated( self, record: UserProfileRecord, facts: tuple[UserProfileFact, ...], ) -> ProfileLifecycleResult: now = utc_now() updated = record.model_copy(update={"facts": facts, "updated_at": now}) self._repository.save(updated) return ProfileLifecycleResult(profile=updated, applied_at=now) @staticmethod def _merge_facts( base: tuple[UserProfileFact, ...], incoming: tuple[UserProfileFact, ...], ) -> tuple[UserProfileFact, ...]: merged: dict[tuple[str, date | None, date | None], UserProfileFact] = { (fact.path, fact.valid_from, fact.valid_to): fact for fact in base } for fact in incoming: merged[(fact.path, fact.valid_from, fact.valid_to)] = fact return tuple(merged.values()) def _emit_event( self, *, event_type: BucketEventType, object_id: str, occurred_at: datetime, payload: dict[str, str] | None = None, ) -> None: payload_body = dict(payload or {}) event = BucketEvent( event_id=derive_bucket_event_id( bucket_id=self._repository.bucket_id, event_type=event_type, occurred_at=occurred_at, actor=_PROFILE_LIFECYCLE_ACTOR, object_type=BucketEventObjectType.PROFILE, object_id=object_id, payload=payload_body, ), bucket_id=self._repository.bucket_id, event_type=event_type, occurred_at=occurred_at, actor=_PROFILE_LIFECYCLE_ACTOR, object_type=BucketEventObjectType.PROFILE, object_id=object_id, payload_version=1, payload=payload_body, ) self._events.save(append_bucket_event(self._events.load(), event)) def _iter_profiles(self) -> Iterable[UserProfileRecord]: """Yield every live profile record by delegating to the repository. Walking the secure-object index lives on :class:`UserProfileLifecycleRepository.iter_records` so the service consumes a public surface instead of reaching for the repository's private secure-object reference. """ return self._repository.iter_records()
def _profile_already_exists_error(*, profile_id: str, bucket_id: str) -> ProfileAlreadyExistsError: return ProfileAlreadyExistsError( _PROFILE_ALREADY_EXISTS_MESSAGE, context={"profile_id": profile_id, "bucket_id": bucket_id}, translated_message="application.user_profile.errors.lifecycle_profile_already_exists", ) def _profile_tombstoned_error(profile_id: str, *, action: Literal["rename", "duplicate"]) -> ProfileNotFoundError: if action == "rename": return ProfileNotFoundError( _PROFILE_TOMBSTONED_RENAME_MESSAGE, context={"profile_id": profile_id, "action": action}, translated_message="application.user_profile.errors.lifecycle_profile_tombstoned_rename", ) if action == "duplicate": return ProfileNotFoundError( _PROFILE_TOMBSTONED_DUPLICATE_MESSAGE, context={"profile_id": profile_id, "action": action}, translated_message="application.user_profile.errors.lifecycle_profile_tombstoned_duplicate", ) def _profile_not_tombstoned_error(profile_id: str) -> ProfileNotFoundError: return ProfileNotFoundError( _PROFILE_NOT_TOMBSTONED_MESSAGE, context={"profile_id": profile_id, "action": "reactivate"}, translated_message="application.user_profile.errors.lifecycle_profile_not_tombstoned", ) __all__ = ["ProfileLifecycleService"]