"""Committed-capsule profile projection repository.
This repository owns only mutable presentation labels. UUID discovery and all
custody facts are delegated to the current-format capsule adapter; a label
record without a valid committed capsule is intentionally invisible.
"""
from __future__ import annotations
from datetime import UTC, datetime
from pathlib import Path
from uuid import UUID
from ...core.paths import effective_storage_root
from ...domain.user_profile.errors import ProfileNotFoundError
from .aggregate import CommittedProfileView, UnlockedProfileFactSummary
from .custody_ports import (
ProfileCustodyCapsuleLabelPort,
ProfileCustodyConcurrentChangeError,
ProfileCustodyLabelHeadPort,
ProfileCustodyPasswordMaterialPort,
ProfileCustodyRecordIntegrityError,
profile_custody_port,
)
from .custody_repository import ProfileCustodyTransactionRepository, profile_custody_transaction_lock
from .custody_transactions import (
ProfileCustodyTransactionConflictError,
ProfileCustodyTransactionCorruptError,
ProfileCustodyTransactionJournal,
ProfileCustodyTransactionOperation,
)
def _require_initial_label_witness(
profile_id: UUID,
journal: ProfileCustodyTransactionJournal,
) -> None:
checks = (
(journal.operation is ProfileCustodyTransactionOperation.CREATE, "operation is not create"),
(journal.profile_id == profile_id, "profile identity differs"),
(journal.label_revision == 1, "label revision is not initial"),
(journal.label_content_digest is not None, "label content digest is absent"),
(journal.label_self_digest is not None, "label self digest is absent"),
)
for valid, reason in checks:
if not valid:
raise ProfileCustodyTransactionConflictError(
f"committed capsule has no canonical initial label witness ({reason})"
)
def _require_initial_label_match(
label_record: ProfileCustodyCapsuleLabelPort,
journal: ProfileCustodyTransactionJournal,
) -> None:
if label_record.label_revision != 1:
return
if any(
(
label_record.content_digest != journal.label_content_digest,
label_record.self_digest != journal.label_self_digest,
label_record.label != journal.label,
)
):
raise ProfileCustodyTransactionConflictError("initial label differs from its create transaction witness")
def _verify_label_head(
*,
root: Path,
label_record: ProfileCustodyCapsuleLabelPort,
source_witness: str,
) -> ProfileCustodyLabelHeadPort:
try:
return profile_custody_port().verify_or_recover_initial_label_head(
label=label_record,
source_witness=source_witness,
root=root,
)
except ProfileCustodyConcurrentChangeError as exc:
raise ProfileCustodyTransactionConflictError(str(exc)) from exc
except ProfileCustodyRecordIntegrityError as exc:
raise ProfileCustodyTransactionCorruptError(str(exc)) from exc
def _load_current_custody_state(
*,
root: Path,
profile_id: UUID,
) -> tuple[ProfileCustodyPasswordMaterialPort, ProfileCustodyCapsuleLabelPort, ProfileCustodyLabelHeadPort]:
with profile_custody_transaction_lock(root, profile_id):
custody = profile_custody_port()
material = custody.load_password_material(profile_id, root=root)
label_record = custody.load_committed_capsule_label(profile_id, root=root)
journal = ProfileCustodyTransactionRepository(root=root).load_journal(material.commit.transaction_id)
_require_initial_label_witness(profile_id, journal)
_require_initial_label_match(label_record, journal)
head = _verify_label_head(root=root, label_record=label_record, source_witness=journal.self_digest)
return material, label_record, head
def _published_at(material: ProfileCustodyPasswordMaterialPort) -> datetime:
return datetime.fromisoformat(material.commit.published_at.replace("Z", "+00:00")).astimezone(UTC)
[docs]
class CommittedProfileRepository:
"""Resolve labels and UUIDs only for validated committed capsules."""
def __init__(self, *, root: Path | None = None) -> None:
"""Bind committed-capsule discovery to the effective storage root."""
self._root = effective_storage_root(root)
@property
def root(self) -> Path:
"""Return the storage root used for committed-capsule discovery."""
return self._root
[docs]
def load(self, profile_id: str | UUID) -> CommittedProfileView:
"""Load one validated committed capsule by canonical UUID."""
try:
identity = UUID(str(profile_id))
except ValueError as exc:
raise ProfileNotFoundError("profile identity is not a canonical UUID") from exc
return self._aggregate_for(identity)
[docs]
def resolve(self, value: str) -> CommittedProfileView:
"""Resolve an exact UUID or exact label without scanning retired buckets."""
try:
return self.load(value)
except ProfileNotFoundError:
pass
matches = [aggregate for aggregate in self.list() if aggregate.label == value]
if len(matches) != 1:
raise ProfileNotFoundError("profile label does not resolve to one committed capsule")
return matches[0]
[docs]
def list(self) -> tuple[CommittedProfileView, ...]:
"""Project only current capsules recognized by the custody adapter."""
result: list[CommittedProfileView] = []
for profile_id in profile_custody_port().list_committed_profile_ids(root=self._root):
try:
result.append(self._aggregate_for(profile_id))
except ProfileNotFoundError:
continue
return tuple(sorted(result, key=lambda item: item.profile_id))
[docs]
def load_unlocked(self, profile_id: str | UUID) -> CommittedProfileView:
"""Project fact state and provenance only through the current authenticated session."""
aggregate = self.load(profile_id)
from ...domain.calculations.registry.authority import bundled_indexed_authority
from .profile_record_repository import ProfileRecordRepository
with bundled_indexed_authority().operation() as operation:
record = ProfileRecordRepository.for_current_session(
aggregate.profile_id,
root=self._root,
profile_decode_context=operation.profile_decode_context(),
).load(aggregate.profile_id)
return aggregate.model_copy(
update={
"fact_summary": UnlockedProfileFactSummary(
setup_state=record.setup_state,
fact_count=len(record.facts),
record_revision=record.record_revision,
content_digest=record.content_digest,
)
}
)
def _aggregate_for(self, profile_id: UUID, *, label_override: str | None = None) -> CommittedProfileView:
if profile_custody_port().committed_capsule_path(profile_id, root=self._root) is None:
raise ProfileNotFoundError("profile has no validated committed custody capsule")
material, label_record, head = _load_current_custody_state(root=self._root, profile_id=profile_id)
if label_override is not None and label_override != label_record.label:
raise ProfileNotFoundError("profile label override differs from its authenticated provenance")
return CommittedProfileView(
profile_id=str(profile_id),
label=label_record.label,
committed_at=_published_at(material),
publication_kind=material.commit.publication_kind,
password_generation=material.envelope.password_generation,
label_revision=head.label_revision,
label_content_digest=head.label_content_digest,
label_self_digest=head.label_self_digest,
label_source_witness=head.self_digest,
)
__all__ = ["CommittedProfileRepository"]