Expand portal onboarding and administration

src/user_engine/service.py

@@ -3,8 +3,8 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field, replace
-from datetime import datetime
-from typing import Any, Iterable, Mapping
+from datetime import datetime, timedelta
+from typing import Any, Callable, Iterable, Mapping
 
 from user_engine.domain import (
     Account,
@@ -325,6 +325,56 @@ class UserEngineService:
             platform_operator=platform_operator,
         )
 
+    def update_self_service_profile(
+        self,
+        actor: Actor,
+        *,
+        display_name: str,
+        consent_accepted: bool,
+        consent_version: str,
+        correlation_id: str | None = None,
+    ) -> User:
+        """Update safe self-service fields; verified email remains immutable."""
+        display_name = display_name.strip()
+        consent_version = consent_version.strip()
+        if not 1 <= len(display_name) <= 200:
+            raise ValidationError("display_name must contain 1 to 200 characters")
+        if consent_accepted and not consent_version:
+            raise ValidationError("consent_version is required when consent is accepted")
+        correlation_id = correlation_id or new_id("corr")
+        identity = self.store.find_identity(actor.issuer, actor.subject)
+        if identity is None:
+            raise NotFoundError("current identity is not linked")
+        user = self._require_user(identity.user_id)
+        decision = self._authorize(
+            actor, action="profile.self.update", resource_type="user-engine:user",
+            resource_id=user.user_id, tenant=actor.tenant,
+            correlation_id=correlation_id, target_user_id=user.user_id,
+        )
+        now = utc_now()
+        updated = replace(
+            user, display_name=display_name, updated_at=now,
+            profile_completed_at=now,
+            consented_at=now if consent_accepted else None,
+            consent_version=consent_version if consent_accepted else None,
+        )
+        with self.store.transaction():
+            self.store.save_user(updated)
+            self._record_mutation(
+                actor, action="profile.self.update", subject=user.user_id,
+                tenant=actor.tenant, correlation_id=correlation_id,
+                decision_id=decision.decision_id,
+                event_type="user.self_service_profile_updated",
+                aggregate_id=user.user_id,
+                payload={
+                    "user_id": user.user_id,
+                    "profile_complete": True,
+                    "consent_accepted": consent_accepted,
+                    "consent_version": consent_version if consent_accepted else None,
+                },
+            )
+        return updated
+
     def start_registration(
         self,
         actor: Actor,
@@ -2362,12 +2412,19 @@ class UserEngineService:
         member: FamilyMemberSpec,
         catalog_namespace: str = "dataspace",
         correlation_id: str | None = None,
+        expires_in: timedelta = timedelta(days=7),
     ) -> FamilyMemberInvitation:
         tenant_context = self.resolve_tenant_context(actor, tenant)
         correlation_id = correlation_id or new_id("corr")
         role = _family_role_value(member.role)
         if not member.primary_email:
             raise ValidationError("family member primary_email is required")
+        if any(
+            item.primary_email.casefold() == member.primary_email.casefold()
+            and item.status == InvitationStatus.PENDING
+            for item in self.store.family_invitations_for_tenant(tenant_context.tenant)
+        ):
+            raise ConflictError("a pending invitation already exists for this address")
         if member.issuer and member.subject:
             existing = self.store.find_identity(member.issuer, member.subject)
             if existing is not None:
@@ -2409,7 +2466,7 @@ class UserEngineService:
                 correlation_id=correlation_id,
             )
             profile_defaults = dict(member.profile_defaults)
-            if member.display_name:
+            if member.display_name and role not in {"user", TENANT_ADMIN_ROLE}:
                 profile_defaults.setdefault("member_display_name", member.display_name)
             self._apply_family_profile_defaults(
                 actor,
@@ -2440,6 +2497,7 @@ class UserEngineService:
                 invited_by=actor.subject,
                 correlation_id=correlation_id,
                 last_sent_correlation_id=correlation_id,
+                expires_at=utc_now() + expires_in,
             )
             self.store.save_family_invitation(invitation)
             self._record_mutation(
@@ -2474,10 +2532,13 @@ class UserEngineService:
         invitation_id: str,
         *,
         correlation_id: str | None = None,
+        expected_version: int | None = None,
     ) -> FamilyInvitation:
         invitation = self._require_family_invitation(invitation_id)
         if invitation.status != InvitationStatus.PENDING:
             raise ValidationError("only pending invitations can be resent")
+        if expected_version is not None and invitation.version != expected_version:
+            raise ConflictError("invitation version does not match")
         tenant_context = self.resolve_tenant_context(actor, invitation.tenant)
         correlation_id = correlation_id or new_id("corr")
         decision = self._authorize(
@@ -2495,6 +2556,8 @@ class UserEngineService:
             resend_count=invitation.resend_count + 1,
             last_sent_correlation_id=correlation_id,
             updated_at=utc_now(),
+            expires_at=utc_now() + timedelta(days=7),
+            version=invitation.version + 1,
         )
         with self.store.transaction():
             self.store.save_family_invitation(updated)
@@ -2522,10 +2585,13 @@ class UserEngineService:
         invitation_id: str,
         *,
         correlation_id: str | None = None,
+        expected_version: int | None = None,
     ) -> FamilyInvitation:
         invitation = self._require_family_invitation(invitation_id)
         if invitation.status != InvitationStatus.PENDING:
             raise ValidationError("only pending invitations can be revoked")
+        if expected_version is not None and invitation.version != expected_version:
+            raise ConflictError("invitation version does not match")
         tenant_context = self.resolve_tenant_context(actor, invitation.tenant)
         correlation_id = correlation_id or new_id("corr")
         decision = self._authorize(
@@ -2551,6 +2617,7 @@ class UserEngineService:
                 status=InvitationStatus.REVOKED,
                 updated_at=utc_now(),
                 revoked_at=utc_now(),
+                version=invitation.version + 1,
             )
             self.store.save_family_invitation(updated)
             self._record_mutation(
@@ -2583,8 +2650,13 @@ class UserEngineService:
             raise ValidationError("revoked invitations cannot be accepted")
         if invitation.status == InvitationStatus.ACCEPTED:
             raise ValidationError("invitation is already accepted")
+        if invitation.expires_at is not None and invitation.expires_at <= utc_now():
+            raise ValidationError("invitation has expired")
         actor = self.identity_adapter.normalize(claims)
         tenant_context = self.resolve_tenant_context(actor, invitation.tenant)
+        claimed_email = str(claims.get("email") or "").strip().casefold()
+        if not claimed_email or claimed_email != invitation.primary_email.casefold():
+            raise AuthorizationDenied("invitation identity does not match")
         correlation_id = correlation_id or new_id("corr")
         decision = self._authorize(
             actor,
@@ -2625,6 +2697,7 @@ class UserEngineService:
                 status=InvitationStatus.ACCEPTED,
                 updated_at=accepted_at,
                 accepted_at=accepted_at,
+                version=invitation.version + 1,
             )
             self.store.save_family_invitation(accepted)
             self._record_mutation(
@@ -2709,6 +2782,73 @@ class UserEngineService:
     def outbox_events(self) -> tuple[OutboxEvent, ...]:
         return self.store.pending_outbox()
 
+    def deliver_outbox(
+        self,
+        actor: Actor,
+        deliver: Callable[[OutboxEvent], None],
+        *,
+        worker_id: str,
+        max_attempts: int = 3,
+        correlation_id: str | None = None,
+    ) -> tuple[OutboxEvent, ...]:
+        """Claim and deliver pending events, retaining bounded failure details."""
+        if not worker_id.strip():
+            raise ValidationError("outbox worker_id is required")
+        if not 1 <= max_attempts <= 20:
+            raise ValidationError("outbox max_attempts must be between 1 and 20")
+        correlation_id = correlation_id or new_id("corr")
+        self._authorize(
+            actor, action="outbox.deliver", resource_type="user-engine:outbox",
+            resource_id="pending", tenant=actor.tenant,
+            correlation_id=correlation_id,
+        )
+        results: list[OutboxEvent] = []
+        for event in self.store.pending_outbox():
+            claimed = replace(
+                event, claimed_by=worker_id, claimed_at=utc_now(),
+                delivery_attempts=event.delivery_attempts + 1,
+                failed_at=None, failure_reason=None,
+            )
+            with self.store.transaction():
+                self.store.save_outbox(claimed)
+            try:
+                deliver(claimed)
+            except Exception as exc:
+                reason = str(exc).strip()[:200] or type(exc).__name__
+                failed = replace(
+                    claimed, failed_at=utc_now(), failure_reason=reason,
+                    dead_lettered_at=(utc_now() if claimed.delivery_attempts >= max_attempts else None),
+                )
+                with self.store.transaction():
+                    self.store.save_outbox(failed)
+                results.append(failed)
+            else:
+                delivered = replace(claimed, delivered_at=utc_now())
+                with self.store.transaction():
+                    self.store.save_outbox(delivered)
+                results.append(delivered)
+        return tuple(results)
+
+    def replay_outbox(
+        self, actor: Actor, event_id: str, *, correlation_id: str | None = None
+    ) -> OutboxEvent:
+        correlation_id = correlation_id or new_id("corr")
+        event = self.store.outbox_event(event_id)
+        if event is None:
+            raise NotFoundError("outbox event not found")
+        self.resolve_tenant_context(actor, event.tenant)
+        self._authorize(
+            actor, action="outbox.replay", resource_type="user-engine:outbox-event",
+            resource_id=event_id, tenant=event.tenant, correlation_id=correlation_id,
+        )
+        replayed = replace(
+            event, claimed_by=None, claimed_at=None, failed_at=None,
+            failure_reason=None, dead_lettered_at=None,
+        )
+        with self.store.transaction():
+            self.store.save_outbox(replayed)
+        return replayed
+
     def outbox_diagnostics(self) -> OutboxDiagnostics:
         event_types: dict[str, int] = {}
         pending = self.store.pending_outbox()
@@ -4728,6 +4868,8 @@ def _family_profile_key(catalog_namespace: str, key: str) -> str:
 
 
 def _family_role_value(role: FamilyRole | str) -> str:
+    if str(role) in {"user", TENANT_ADMIN_ROLE}:
+        return str(role)
     try:
         return FamilyRole(str(role)).value
     except ValueError as exc: