From 28b60782de0902448519bbcf4f3ebdfb5a9183f8 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Thu, 30 Jul 2026 04:00:31 +0200 Subject: [PATCH] feat: route bounded hierarchy postbox copies --- README.md | 12 +- docs/POSTBOX_CONCEPT.md | 18 + src/govoplan_postbox/backend/db/models.py | 20 + .../backend/hierarchy_routing.py | 348 ++++++ src/govoplan_postbox/backend/manifest.py | 14 +- .../e4b7c9d2a6f1_v011_hierarchy_routes.py | 54 + src/govoplan_postbox/backend/router.py | 33 +- src/govoplan_postbox/backend/schemas.py | 165 ++- src/govoplan_postbox/backend/service.py | 1008 ++++++++++++++++- tests/test_manifest.py | 2 + tests/test_migration.py | 20 + tests/test_router.py | 17 + tests/test_service.py | 630 +++++++++++ webui/src/api/postbox.ts | 56 +- .../features/postbox/PostboxAdminPanel.tsx | 305 ++++- webui/src/styles/postbox.css | 30 + 16 files changed, 2715 insertions(+), 17 deletions(-) create mode 100644 src/govoplan_postbox/backend/hierarchy_routing.py create mode 100644 src/govoplan_postbox/backend/migrations/versions/e4b7c9d2a6f1_v011_hierarchy_routes.py diff --git a/README.md b/README.md index fbf0523..5cdf4e6 100644 --- a/README.md +++ b/README.md @@ -80,7 +80,17 @@ addresses, exact function-bound Postboxes, current IDM assignment access decisions, vacancy status, idempotent producer delivery, source-preserving message and attachment references, personal read/acknowledgement receipts, unified inbox projections, access evidence, an inbox route, and tenant -administration. +administration. Published template revisions can also opt into bounded linked +copies through one explicit organization structure. Classification, producer, +retention, stop, depth, target-template, and target-function gates are frozen +at delivery time and exposed through delivery evidence and the routing dry-run +API. + +Vacancy escalation is a separate attention policy. It creates no personal +account grant: a durable route waits for its configured delay and then creates +an independently readable copy in the next frozen function Postbox. The +`govoplan.postbox.dispatch_routes` periodic Core worker drains due routes when +Celery beat and a worker consuming the `postbox` queue are enabled. The persistence model reserves ciphertext manifests, wrapped keys, key epochs, expiry, and withdrawal state. The active profile remains `plaintext_v1`; the diff --git a/docs/POSTBOX_CONCEPT.md b/docs/POSTBOX_CONCEPT.md index de5d1c8..0c7e764 100644 --- a/docs/POSTBOX_CONCEPT.md +++ b/docs/POSTBOX_CONCEPT.md @@ -176,6 +176,24 @@ Vacancy is a visible delivery/attention state rather than an automatic grant to an unrelated personal account. Policy may trigger a bounded escalation after a delay. +The implemented policy keeps the three semantics separate: + +- `linked_copy` can target the nearest matching ancestor or every bounded + matching ancestor in one selected structure. +- `attention` currently supports delayed vacancy escalation over the remaining + delivery-time target snapshot. +- `shared_visibility` remains explicitly disabled until its access and + encryption semantics are implemented. + +Routing is off unless an immutable template revision enables it and supplies a +target template, target function type, producer allowlist, classification +allowlist, depth, and structure. Optional relation, stop-unit, stop-unit-type, +expiry, and maximum-retention gates narrow the route further. The dry-run API +returns blocked and unavailable candidates without materializing addresses. +Delivery materializes only frozen candidates, stores path-edge provenance, and +creates source-preserving copies with independent read and acknowledgement +receipts. + ## Campaign Distribution Campaign can use Postbox as an explicit delivery channel through diff --git a/src/govoplan_postbox/backend/db/models.py b/src/govoplan_postbox/backend/db/models.py index 5de8b8b..4a96c1b 100644 --- a/src/govoplan_postbox/backend/db/models.py +++ b/src/govoplan_postbox/backend/db/models.py @@ -641,8 +641,20 @@ class PostboxDelivery(Base, TimestampMixin): class PostboxRoute(Base, TimestampMixin): __tablename__ = "postbox_routes" __table_args__ = ( + UniqueConstraint( + "delivery_id", + "target_postbox_id", + "route_kind", + "depth", + name="uq_postbox_routes_delivery_target_kind_depth", + ), Index("ix_postbox_routes_delivery_kind", "delivery_id", "route_kind"), Index("ix_postbox_routes_target", "tenant_id", "target_postbox_id"), + Index( + "ix_postbox_routes_due", + "status", + "execute_after", + ), ) id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) @@ -676,6 +688,14 @@ class PostboxRoute(Base, TimestampMixin): ForeignKey("postbox_routes.id", ondelete="SET NULL"), nullable=True, ) + execute_after: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + processed_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) policy_snapshot: Mapped[dict[str, Any]] = mapped_column( JSON, default=dict, diff --git a/src/govoplan_postbox/backend/hierarchy_routing.py b/src/govoplan_postbox/backend/hierarchy_routing.py new file mode 100644 index 0000000..fa4c0c9 --- /dev/null +++ b/src/govoplan_postbox/backend/hierarchy_routing.py @@ -0,0 +1,348 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any, Mapping + +from govoplan_core.core.idm import ( + IdmFunctionAssignmentDirectory, + OrganizationFunctionAssignmentRef, +) +from govoplan_core.core.organizations import ( + OrganizationFunctionRef, + OrganizationHierarchyDirectory, + OrganizationHierarchyEdgeRef, + OrganizationUnitRef, +) +from govoplan_postbox.backend.schemas import PostboxRoutingPolicyPayload + + +@dataclass(frozen=True, slots=True) +class HierarchyRouteCandidate: + depth: int + unit: OrganizationUnitRef + function: OrganizationFunctionRef | None + holders: tuple[OrganizationFunctionAssignmentRef, ...] + status: str + path: tuple[Mapping[str, object], ...] + diagnostics: tuple[str, ...] = () + + @property + def holder_count(self) -> int: + return len( + { + holder.identity_id or holder.account_id or holder.id + for holder in self.holders + } + ) + + +@dataclass(frozen=True, slots=True) +class HierarchyRoutePlan: + status: str + policy: Mapping[str, object] + candidates: tuple[HierarchyRouteCandidate, ...] = () + diagnostics: tuple[str, ...] = () + + +def normalized_routing_policy( + value: Mapping[str, object] | None, +) -> dict[str, object]: + return PostboxRoutingPolicyPayload.model_validate(value or {}).model_dump( + mode="json" + ) + + +def plan_hierarchy_routes( + *, + hierarchy: OrganizationHierarchyDirectory | None, + incumbencies: IdmFunctionAssignmentDirectory, + tenant_id: str, + source_unit_id: str | None, + routing_policy: Mapping[str, object] | None, + producer_module: str, + classification: str, + expires_at: datetime | None, + now: datetime, +) -> HierarchyRoutePlan: + policy = normalized_routing_policy(routing_policy) + linked_copy = _mapping(policy.get("linked_copy")) + if not linked_copy.get("enabled"): + return HierarchyRoutePlan( + status="disabled", + policy=policy, + diagnostics=("hierarchy_routing_disabled",), + ) + if hierarchy is None: + return HierarchyRoutePlan( + status="blocked", + policy=policy, + diagnostics=("organization_hierarchy_unavailable",), + ) + if not source_unit_id: + return HierarchyRoutePlan( + status="blocked", + policy=policy, + diagnostics=("source_organization_unit_missing",), + ) + + diagnostics = list( + _policy_gate_diagnostics( + linked_copy, + producer_module=producer_module, + classification=classification, + expires_at=expires_at, + now=now, + ) + ) + if diagnostics: + return HierarchyRoutePlan( + status="blocked", + policy=policy, + diagnostics=tuple(diagnostics), + ) + + structure_id = str(linked_copy["structure_id"]) + relation_type_ids = tuple( + str(value) for value in linked_copy.get("relation_type_ids", ()) + ) + max_depth = int(linked_copy["max_depth"]) + try: + resolutions = hierarchy.resolve_hierarchy_relatives( + tenant_id, + (source_unit_id,), + structure_id=structure_id, + relation_type_ids=relation_type_ids, + direction="ancestors", + max_depth=max_depth, + ) + except ValueError as exc: + return HierarchyRoutePlan( + status="blocked", + policy=policy, + diagnostics=(f"hierarchy_request_invalid:{exc}",), + ) + if not resolutions: + return HierarchyRoutePlan( + status="blocked", + policy=policy, + diagnostics=("hierarchy_resolution_missing",), + ) + resolution = resolutions[0] + diagnostics.extend(resolution.diagnostics) + if resolution.cycle_detected: + diagnostics.append("hierarchy_cycle_bounded") + if resolution.depth_limited: + diagnostics.append("hierarchy_depth_limited") + if resolution.status != "active": + diagnostics.append(f"hierarchy_{resolution.status}") + return HierarchyRoutePlan( + status="blocked", + policy=policy, + diagnostics=tuple(dict.fromkeys(diagnostics)), + ) + + matches = [] + stop_unit_id = linked_copy.get("stop_unit_id") + stop_unit_type_id = linked_copy.get("stop_unit_type_id") + for match in sorted( + resolution.matches, + key=lambda item: (item.depth, item.unit.name, item.unit.id), + ): + matches.append(match) + if ( + stop_unit_id + and match.unit.id == stop_unit_id + or stop_unit_type_id + and match.unit.unit_type_id == stop_unit_type_id + ): + diagnostics.append("hierarchy_stop_reached") + break + if not matches: + return HierarchyRoutePlan( + status="no_route", + policy=policy, + diagnostics=tuple( + dict.fromkeys((*diagnostics, "no_hierarchy_ancestor")) + ), + ) + + unit_ids = tuple(dict.fromkeys(match.unit.id for match in matches)) + target_function_type_id = str(linked_copy["target_function_type_id"]) + try: + function_resolution = hierarchy.resolve_functions_by_type( + tenant_id, + target_function_type_id, + organization_unit_ids=unit_ids, + ) + except ValueError as exc: + return HierarchyRoutePlan( + status="blocked", + policy=policy, + diagnostics=tuple( + dict.fromkeys( + (*diagnostics, f"function_resolution_invalid:{exc}") + ) + ), + ) + diagnostics.extend(function_resolution.diagnostics) + if function_resolution.status != "active": + diagnostics.append( + f"target_function_type_{function_resolution.status}" + ) + + functions_by_unit: dict[str, list[OrganizationFunctionRef]] = {} + for function in function_resolution.matches: + if function.status != "active": + continue + functions_by_unit.setdefault( + function.organization_unit_id, + [], + ).append(function) + function_ids = tuple( + function.id + for functions in functions_by_unit.values() + if len(functions) == 1 + for function in functions + ) + try: + holder_map = ( + incumbencies.organization_function_incumbencies( + function_ids, + tenant_id=tenant_id, + ) + if function_ids + else {} + ) + except ValueError: + holder_map = {} + diagnostics.append("target_incumbency_unavailable") + + candidates: list[HierarchyRouteCandidate] = [] + seen_targets: set[tuple[str, str]] = set() + for match in matches: + functions = functions_by_unit.get(match.unit.id, []) + candidate_diagnostics: list[str] = [] + function = functions[0] if len(functions) == 1 else None + if match.unit.status != "active": + status = "unit_inactive" + candidate_diagnostics.append("target_unit_inactive") + elif not functions: + status = "function_missing" + candidate_diagnostics.append("target_function_missing") + elif len(functions) > 1: + status = "function_ambiguous" + candidate_diagnostics.append("target_function_ambiguous") + elif (match.unit.id, function.id) in seen_targets: + status = "duplicate" + candidate_diagnostics.append("duplicate_target_suppressed") + else: + seen_targets.add((match.unit.id, function.id)) + holders = tuple( + holder_map.get(function.id).assignments + if function.id in holder_map + else () + ) + status = "available" if holders else "vacant" + candidates.append( + HierarchyRouteCandidate( + depth=match.depth, + unit=match.unit, + function=function, + holders=holders, + status=status, + path=tuple(_edge_snapshot(edge) for edge in match.path), + ) + ) + continue + candidates.append( + HierarchyRouteCandidate( + depth=match.depth, + unit=match.unit, + function=function, + holders=(), + status=status, + path=tuple(_edge_snapshot(edge) for edge in match.path), + diagnostics=tuple(candidate_diagnostics), + ) + ) + + routable = [ + candidate + for candidate in candidates + if candidate.status in {"available", "vacant"} + ] + status = "planned" if routable else "no_route" + if not routable: + diagnostics.append("no_available_hierarchy_target") + return HierarchyRoutePlan( + status=status, + policy=policy, + candidates=tuple(candidates), + diagnostics=tuple(dict.fromkeys(diagnostics)), + ) + + +def _policy_gate_diagnostics( + linked_copy: Mapping[str, object], + *, + producer_module: str, + classification: str, + expires_at: datetime | None, + now: datetime, +) -> tuple[str, ...]: + diagnostics: list[str] = [] + classifications = { + str(value) for value in linked_copy.get("allowed_classifications", ()) + } + if classification not in classifications: + diagnostics.append("classification_not_allowed") + producers = { + str(value) for value in linked_copy.get("allowed_producer_modules", ()) + } + if producer_module not in producers and "*" not in producers: + diagnostics.append("producer_not_authorized") + normalized_expiry = _as_utc(expires_at) if expires_at else None + if linked_copy.get("require_expiry") and normalized_expiry is None: + diagnostics.append("expiry_required") + max_retention_days = linked_copy.get("max_retention_days") + if ( + normalized_expiry is not None + and max_retention_days is not None + and normalized_expiry + > _as_utc(now) + timedelta(days=int(max_retention_days)) + ): + diagnostics.append("retention_limit_exceeded") + if normalized_expiry is not None and normalized_expiry <= _as_utc(now): + diagnostics.append("message_already_expired") + return tuple(diagnostics) + + +def _edge_snapshot(edge: OrganizationHierarchyEdgeRef) -> dict[str, object]: + return { + "edge_id": edge.id, + "structure_id": edge.structure.id, + "structure_slug": edge.structure.slug, + "relation_type_id": edge.relation_type.id, + "relation_type_slug": edge.relation_type.slug, + "source_unit_id": edge.source_unit_id, + "target_unit_id": edge.target_unit_id, + } + + +def _mapping(value: object) -> dict[str, Any]: + return dict(value) if isinstance(value, Mapping) else {} + + +def _as_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +__all__ = [ + "HierarchyRouteCandidate", + "HierarchyRoutePlan", + "normalized_routing_policy", + "plan_hierarchy_routes", +] diff --git a/src/govoplan_postbox/backend/manifest.py b/src/govoplan_postbox/backend/manifest.py index 0dc4b47..460482a 100644 --- a/src/govoplan_postbox/backend/manifest.py +++ b/src/govoplan_postbox/backend/manifest.py @@ -29,13 +29,17 @@ from govoplan_core.core.modules import ( RoleTemplate, ) from govoplan_core.core.notifications import CAPABILITY_NOTIFICATIONS_DISPATCH -from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY +from govoplan_core.core.organizations import ( + CAPABILITY_ORGANIZATION_DIRECTORY, + CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY, +) from govoplan_core.core.postbox import ( CAPABILITY_POSTBOX_ACCESS, CAPABILITY_POSTBOX_DELIVERY, CAPABILITY_POSTBOX_DIRECTORY, CAPABILITY_POSTBOX_EVIDENCE, CAPABILITY_POSTBOX_MESSAGES, + CAPABILITY_POSTBOX_ROUTING, ) from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base @@ -201,6 +205,7 @@ manifest = ModuleManifest( CAPABILITY_IDM_DIRECTORY, CAPABILITY_IDM_FUNCTION_ASSIGNMENTS, CAPABILITY_ORGANIZATION_DIRECTORY, + CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY, ), provides_interfaces=tuple( ModuleInterfaceProvider(name=name, version=MODULE_VERSION) @@ -210,6 +215,7 @@ manifest = ModuleManifest( CAPABILITY_POSTBOX_MESSAGES, CAPABILITY_POSTBOX_DELIVERY, CAPABILITY_POSTBOX_EVIDENCE, + CAPABILITY_POSTBOX_ROUTING, ) ), requires_interfaces=( @@ -218,6 +224,11 @@ manifest = ModuleManifest( version_min="0.1.8", version_max_exclusive="0.2.0", ), + ModuleInterfaceRequirement( + name="organizations.hierarchy_directory", + version_min="0.1.0", + version_max_exclusive="0.2.0", + ), ModuleInterfaceRequirement( name=CAPABILITY_NOTIFICATIONS_DISPATCH, version_min="0.1.8", @@ -318,6 +329,7 @@ manifest = ModuleManifest( CAPABILITY_POSTBOX_MESSAGES: _configure, CAPABILITY_POSTBOX_DELIVERY: _configure, CAPABILITY_POSTBOX_EVIDENCE: _configure, + CAPABILITY_POSTBOX_ROUTING: _configure, }, documentation=( DocumentationTopic( diff --git a/src/govoplan_postbox/backend/migrations/versions/e4b7c9d2a6f1_v011_hierarchy_routes.py b/src/govoplan_postbox/backend/migrations/versions/e4b7c9d2a6f1_v011_hierarchy_routes.py new file mode 100644 index 0000000..54ece0d --- /dev/null +++ b/src/govoplan_postbox/backend/migrations/versions/e4b7c9d2a6f1_v011_hierarchy_routes.py @@ -0,0 +1,54 @@ +"""v0.1.1 durable hierarchy routes + +Revision ID: e4b7c9d2a6f1 +Revises: c7d2e5f8a1b4 +Create Date: 2026-07-30 04:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "e4b7c9d2a6f1" +down_revision = "c7d2e5f8a1b4" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("postbox_routes") as batch_op: + batch_op.add_column( + sa.Column( + "execute_after", + sa.DateTime(timezone=True), + nullable=True, + ) + ) + batch_op.add_column( + sa.Column( + "processed_at", + sa.DateTime(timezone=True), + nullable=True, + ) + ) + batch_op.create_unique_constraint( + "uq_postbox_routes_delivery_target_kind_depth", + ("delivery_id", "target_postbox_id", "route_kind", "depth"), + ) + batch_op.create_index( + "ix_postbox_routes_due", + ("status", "execute_after"), + unique=False, + ) + + +def downgrade() -> None: + with op.batch_alter_table("postbox_routes") as batch_op: + batch_op.drop_index("ix_postbox_routes_due") + batch_op.drop_constraint( + "uq_postbox_routes_delivery_target_kind_depth", + type_="unique", + ) + batch_op.drop_column("processed_at") + batch_op.drop_column("execute_after") diff --git a/src/govoplan_postbox/backend/router.py b/src/govoplan_postbox/backend/router.py index 84e5843..8409872 100644 --- a/src/govoplan_postbox/backend/router.py +++ b/src/govoplan_postbox/backend/router.py @@ -39,6 +39,8 @@ from govoplan_postbox.backend.schemas import ( PostboxMessageListResponse, PostboxMessageStateRequest, PostboxOrganizationTargetsResponse, + PostboxRouteDryRunRequest, + PostboxRouteDryRunResponse, PostboxTemplateCreateRequest, PostboxTemplateItem, PostboxTemplateListResponse, @@ -395,6 +397,30 @@ def api_deliver_to_postbox( return PostboxDeliveryResponse.model_validate(asdict(result)) +@router.post( + "/routing/dry-run", + response_model=PostboxRouteDryRunResponse, +) +def api_preview_postbox_routing( + payload: PostboxRouteDryRunRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PostboxRouteDryRunResponse: + _require_any(principal, DELIVERY_SCOPE, TEMPLATE_ADMIN_SCOPE) + try: + result = get_service().preview_hierarchy_routes( + session, + tenant_id=principal.tenant_id, + target=PostboxTargetRef(**payload.target.model_dump()), + producer_module=payload.producer_module, + classification=payload.classification, + expires_at=payload.expires_at, + ) + except PostboxError as exc: + raise _http_error(exc) from exc + return PostboxRouteDryRunResponse.model_validate(result) + + @router.get("/groupings", response_model=PostboxGroupingListResponse) def api_list_postbox_groupings( session: Session = Depends(get_session), @@ -502,7 +528,12 @@ def api_postbox_organization_targets( return PostboxOrganizationTargetsResponse( units=list( get_service().organization_targets(tenant_id=principal.tenant_id) - ) + ), + structures=list( + get_service().organization_hierarchy_targets( + tenant_id=principal.tenant_id + ) + ), ) diff --git a/src/govoplan_postbox/backend/schemas.py b/src/govoplan_postbox/backend/schemas.py index 6930330..910d139 100644 --- a/src/govoplan_postbox/backend/schemas.py +++ b/src/govoplan_postbox/backend/schemas.py @@ -150,6 +150,144 @@ class PostboxDeliveryResponse(BaseModel): evidence: dict[str, Any] = Field(default_factory=dict) +class PostboxLinkedCopyPolicyPayload(BaseModel): + enabled: bool = False + structure_id: str | None = Field(default=None, max_length=36) + relation_type_ids: list[str] = Field(default_factory=list, max_length=20) + max_depth: int = Field(default=1, ge=1, le=20) + stop_unit_id: str | None = Field(default=None, max_length=36) + stop_unit_type_id: str | None = Field(default=None, max_length=36) + target_function_type_id: str | None = Field(default=None, max_length=36) + target_template_id: str | None = Field(default=None, max_length=36) + fanout: Literal["nearest", "all"] = "nearest" + allowed_classifications: list[str] = Field( + default_factory=lambda: ["internal"], + max_length=20, + ) + allowed_producer_modules: list[str] = Field( + default_factory=list, + max_length=50, + ) + require_expiry: bool = False + max_retention_days: int | None = Field(default=None, ge=1, le=36500) + + @model_validator(mode="after") + def validate_enabled_policy(self) -> "PostboxLinkedCopyPolicyPayload": + self.relation_type_ids = list(dict.fromkeys(self.relation_type_ids)) + self.allowed_classifications = list( + dict.fromkeys( + value.strip() for value in self.allowed_classifications + ) + ) + self.allowed_producer_modules = list( + dict.fromkeys( + value.strip() for value in self.allowed_producer_modules + ) + ) + if any(not value for value in self.relation_type_ids): + raise ValueError("Relation type IDs must not be empty.") + if any(not value for value in self.allowed_classifications): + raise ValueError("Allowed classifications must not be empty.") + if any(not value for value in self.allowed_producer_modules): + raise ValueError("Allowed producer modules must not be empty.") + if self.enabled and not all( + ( + self.structure_id, + self.target_function_type_id, + self.target_template_id, + self.allowed_classifications, + self.allowed_producer_modules, + ) + ): + raise ValueError( + "Enabled hierarchy copy requires a structure, target function " + "type, target template, classification gate, and producer allowlist." + ) + return self + + +class PostboxAttentionPolicyPayload(BaseModel): + mode: Literal["none", "vacancy_escalation"] = "none" + delay_minutes: int | None = Field(default=None, ge=1, le=43200) + + @model_validator(mode="after") + def validate_delay(self) -> "PostboxAttentionPolicyPayload": + if self.mode == "vacancy_escalation" and self.delay_minutes is None: + raise ValueError("Vacancy escalation requires a delay.") + if self.mode == "none": + self.delay_minutes = None + return self + + +class PostboxSharedVisibilityPolicyPayload(BaseModel): + mode: Literal["none"] = "none" + + +class PostboxRoutingPolicyPayload(BaseModel): + linked_copy: PostboxLinkedCopyPolicyPayload = Field( + default_factory=PostboxLinkedCopyPolicyPayload + ) + attention: PostboxAttentionPolicyPayload = Field( + default_factory=PostboxAttentionPolicyPayload + ) + shared_visibility: PostboxSharedVisibilityPolicyPayload = Field( + default_factory=PostboxSharedVisibilityPolicyPayload + ) + + @model_validator(mode="before") + @classmethod + def normalize_legacy_policy(cls, value: Any) -> Any: + if value in (None, {}, {"mode": "none"}): + return {} + return value + + @model_validator(mode="after") + def validate_semantics(self) -> "PostboxRoutingPolicyPayload": + if ( + self.attention.mode == "vacancy_escalation" + and ( + not self.linked_copy.enabled + or self.linked_copy.fanout != "nearest" + ) + ): + raise ValueError( + "Vacancy escalation requires nearest linked-copy routing." + ) + return self + + +class PostboxRoutePreviewTarget(BaseModel): + depth: int + organization_unit_id: str + organization_unit_name: str + function_id: str | None = None + function_name: str | None = None + target_postbox_id: str | None = None + target_address: str | None = None + status: str + vacant: bool = True + holder_count: int = 0 + path: list[dict[str, Any]] = Field(default_factory=list) + diagnostics: list[str] = Field(default_factory=list) + + +class PostboxRouteDryRunRequest(BaseModel): + target: PostboxTargetPayload + producer_module: str = Field(min_length=1, max_length=100) + classification: str = Field(default="internal", min_length=1, max_length=50) + expires_at: datetime | None = None + + +class PostboxRouteDryRunResponse(BaseModel): + status: str + source_postbox_id: str | None = None + policy: PostboxRoutingPolicyPayload = Field( + default_factory=PostboxRoutingPolicyPayload + ) + routes: list[PostboxRoutePreviewTarget] = Field(default_factory=list) + diagnostics: list[str] = Field(default_factory=list) + + class PostboxExactCreateRequest(BaseModel): name: str = Field(min_length=1, max_length=500) description: str | None = None @@ -175,6 +313,9 @@ class PostboxTemplateRevisionPayload(BaseModel): ) classification: str = Field(default="internal", min_length=1, max_length=50) allow_vacant_delivery: bool = True + routing_policy: PostboxRoutingPolicyPayload = Field( + default_factory=PostboxRoutingPolicyPayload + ) class PostboxTemplateCreateRequest(PostboxTemplateRevisionPayload): @@ -188,7 +329,6 @@ class PostboxTemplateRevisionItem(PostboxTemplateRevisionPayload): revision: int encryption_profile: str history_policy: dict[str, Any] = Field(default_factory=dict) - routing_policy: dict[str, Any] = Field(default_factory=dict) retention_policy: dict[str, Any] = Field(default_factory=dict) published_at: datetime | None = None created_at: datetime @@ -240,8 +380,31 @@ class PostboxOrganizationUnitItem(BaseModel): functions: list[PostboxOrganizationFunctionItem] = Field(default_factory=list) +class PostboxOrganizationRelationTypeItem(BaseModel): + id: str + slug: str + name: str + structure_id: str | None = None + is_hierarchical: bool = True + status: str = "active" + + +class PostboxOrganizationStructureItem(BaseModel): + id: str + slug: str + name: str + structure_kind: str + status: str = "active" + relation_types: list[PostboxOrganizationRelationTypeItem] = Field( + default_factory=list + ) + + class PostboxOrganizationTargetsResponse(BaseModel): units: list[PostboxOrganizationUnitItem] + structures: list[PostboxOrganizationStructureItem] = Field( + default_factory=list + ) class PostboxGroupingPayload(BaseModel): diff --git a/src/govoplan_postbox/backend/service.py b/src/govoplan_postbox/backend/service.py index 343082f..cd402d0 100644 --- a/src/govoplan_postbox/backend/service.py +++ b/src/govoplan_postbox/backend/service.py @@ -4,11 +4,12 @@ import hashlib import logging import re from collections.abc import Mapping, Sequence +from datetime import datetime, timedelta, timezone from typing import Literal from sqlalchemy import and_, func, or_ from sqlalchemy.exc import IntegrityError -from sqlalchemy.orm import Session, selectinload +from sqlalchemy.orm import Session, object_session, selectinload from govoplan_core.core.events import ( EventActorRef, @@ -35,8 +36,10 @@ from govoplan_core.core.notifications import ( ) from govoplan_core.core.organizations import ( CAPABILITY_ORGANIZATION_DIRECTORY, + CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY, OrganizationDirectory, OrganizationFunctionRef, + OrganizationHierarchyDirectory, OrganizationUnitRef, ) from govoplan_core.core.postbox import ( @@ -71,9 +74,16 @@ from govoplan_postbox.backend.db.models import ( PostboxMessage, PostboxMessageReceipt, PostboxParticipant, + PostboxRoute, PostboxTemplate, PostboxTemplateRevision, ) +from govoplan_postbox.backend.hierarchy_routing import ( + HierarchyRouteCandidate, + HierarchyRoutePlan, + normalized_routing_policy, + plan_hierarchy_routes, +) logger = logging.getLogger(__name__) @@ -107,6 +117,12 @@ def _mapping(value: Mapping[str, object] | None) -> dict[str, object]: return dict(value or {}) +def _as_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + def _publish_postbox_event( session: Session, event_type: str, @@ -150,12 +166,14 @@ class PostboxService: idm: IdmDirectory, incumbencies: IdmFunctionAssignmentDirectory, organizations: OrganizationDirectory, + hierarchy: OrganizationHierarchyDirectory | None = None, notifications: NotificationDispatchProvider | None = None, ) -> None: self._identities = identities self._idm = idm self._incumbencies = incumbencies self._organizations = organizations + self._hierarchy = hierarchy self._notifications = notifications @classmethod @@ -168,6 +186,9 @@ class PostboxService: organizations = registry.require_capability( CAPABILITY_ORGANIZATION_DIRECTORY ) + hierarchy = registry.require_capability( + CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY + ) if not isinstance(identities, IdentityDirectory): raise RuntimeError( f"Invalid capability: {CAPABILITY_IDENTITY_DIRECTORY}" @@ -182,11 +203,17 @@ class PostboxService: raise RuntimeError( f"Invalid capability: {CAPABILITY_ORGANIZATION_DIRECTORY}" ) + if not isinstance(hierarchy, OrganizationHierarchyDirectory): + raise RuntimeError( + "Invalid capability: " + f"{CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY}" + ) return cls( identities=identities, idm=idm, incumbencies=incumbencies, organizations=organizations, + hierarchy=hierarchy, notifications=notification_dispatch_provider(registry), ) @@ -841,6 +868,14 @@ class PostboxService: "delivery_id": delivery.id, }, ) + routes = self._apply_hierarchy_routing( + db, + request=request, + delivery=delivery, + source_postbox=postbox, + source_message=message, + source_revision=revision, + ) _publish_postbox_event( db, "postbox.delivery.accepted.v1", @@ -855,6 +890,8 @@ class PostboxService: "status": delivery.status, "vacant": not bool(holders), "holder_count": delivery.holder_count, + "route_count": len(routes), + "route_ids": [route.id for route in routes], "producer_resource_type": request.producer_resource_type, "producer_resource_id": request.producer_resource_id, }, @@ -868,6 +905,906 @@ class PostboxService: ) return self._delivery_result(delivery, postbox) + def preview_hierarchy_routes( + self, + session: Session, + *, + tenant_id: str, + target: PostboxTargetRef, + producer_module: str, + classification: str, + expires_at: datetime | None, + ) -> dict[str, object]: + entry = self.resolve_postbox( + session, + tenant_id=tenant_id, + target=target, + materialize=False, + ) + if entry is None: + raise PostboxError( + "target_not_materialized", + "Dry-run routing requires an existing source Postbox.", + ) + source_postbox = self._get_postbox( + session, + tenant_id=tenant_id, + postbox_id=entry.id, + ) + source_revision = self._address_revision( + session, + source_postbox.address_record, + ) + plan = self._hierarchy_route_plan( + source_postbox=source_postbox, + source_revision=source_revision, + producer_module=producer_module, + classification=classification, + expires_at=expires_at, + ) + target_template, target_revision, template_diagnostics = ( + self._route_target_template( + session, + tenant_id=tenant_id, + plan=plan, + required=False, + ) + ) + routes = [ + self._route_preview( + session, + tenant_id=tenant_id, + source_postbox=source_postbox, + candidate=candidate, + target_template=target_template, + target_revision=target_revision, + ) + for candidate in plan.candidates + ] + return { + "status": ( + "blocked" + if template_diagnostics and plan.status == "planned" + else plan.status + ), + "source_postbox_id": source_postbox.id, + "policy": dict(plan.policy), + "routes": routes, + "diagnostics": list( + dict.fromkeys((*plan.diagnostics, *template_diagnostics)) + ), + } + + def dispatch_due_routes( + self, + session: object, + *, + tenant_id: str | None = None, + limit: int = 50, + ) -> Mapping[str, object]: + db = _session(session) + bounded_limit = max(1, min(int(limit), 500)) + now = utc_now() + query = db.query(PostboxRoute).filter( + PostboxRoute.status == "pending_vacancy_escalation", + PostboxRoute.execute_after.is_not(None), + PostboxRoute.execute_after <= now, + ) + if tenant_id: + query = query.filter(PostboxRoute.tenant_id == tenant_id) + routes = ( + query.order_by( + PostboxRoute.execute_after, + PostboxRoute.created_at, + PostboxRoute.id, + ) + .with_for_update() + .limit(bounded_limit) + .all() + ) + result = { + "selected": len(routes), + "delivered": 0, + "vacant": 0, + "rescheduled": 0, + "cancelled": 0, + "failed": 0, + "route_ids": [], + } + for route in routes: + result["route_ids"].append(route.id) + try: + with db.begin_nested(): + outcome, rescheduled = self._dispatch_pending_route( + db, + route=route, + now=now, + ) + result[outcome] += 1 + if rescheduled: + result["rescheduled"] += 1 + except PostboxError as exc: + failed_route = db.get(PostboxRoute, route.id) + if failed_route is not None: + failed_route.status = f"failed:{exc.code}"[:40] + failed_route.processed_at = now + result["failed"] += 1 + except Exception: # noqa: BLE001 - preserve other due routes. + logger.exception( + "Postbox hierarchy route dispatch failed", + extra={"postbox_route_id": route.id}, + ) + failed_route = db.get(PostboxRoute, route.id) + if failed_route is not None: + failed_route.status = "failed:unexpected" + failed_route.processed_at = now + result["failed"] += 1 + db.flush() + return result + + def _hierarchy_route_plan( + self, + *, + source_postbox: Postbox, + source_revision: PostboxTemplateRevision | None, + producer_module: str, + classification: str, + expires_at: datetime | None, + ) -> HierarchyRoutePlan: + routing_policy = ( + source_revision.routing_policy + if source_revision is not None + else _mapping( + (source_postbox.settings or {}).get("routing_policy") + if isinstance(source_postbox.settings, Mapping) + else None + ) + ) + return plan_hierarchy_routes( + hierarchy=self._hierarchy, + incumbencies=self._incumbencies, + tenant_id=source_postbox.tenant_id, + source_unit_id=source_postbox.address_record.organization_unit_id, + routing_policy=routing_policy, + producer_module=producer_module, + classification=classification, + expires_at=expires_at, + now=utc_now(), + ) + + def _route_target_template( + self, + session: Session, + *, + tenant_id: str, + plan: HierarchyRoutePlan, + required: bool, + ) -> tuple[ + PostboxTemplate | None, + PostboxTemplateRevision | None, + tuple[str, ...], + ]: + linked_copy = _mapping(plan.policy.get("linked_copy")) + template_id = str(linked_copy.get("target_template_id") or "") + if not template_id: + return None, None, ( + ("target_template_missing",) + if required or plan.status == "planned" + else () + ) + try: + template = self._get_template( + session, + tenant_id=tenant_id, + template_id=template_id, + ) + except PostboxError: + return None, None, ("target_template_missing",) + if template.status != "published" or not template.published_revision_id: + return template, None, ("target_template_not_published",) + revision = next( + ( + item + for item in template.revisions + if item.id == template.published_revision_id + ), + None, + ) + if revision is None: + return template, None, ("target_template_revision_missing",) + target_function_type_id = linked_copy.get("target_function_type_id") + if ( + revision.function_type_id + and revision.function_type_id != target_function_type_id + ): + return template, revision, ("target_template_function_mismatch",) + return template, revision, () + + def _route_preview( + self, + session: Session, + *, + tenant_id: str, + source_postbox: Postbox, + candidate: HierarchyRouteCandidate, + target_template: PostboxTemplate | None, + target_revision: PostboxTemplateRevision | None, + ) -> dict[str, object]: + diagnostics = list(candidate.diagnostics) + status = candidate.status + target_postbox: Postbox | None = None + if ( + candidate.function is not None + and target_template is not None + and target_revision is not None + ): + key = self._template_address_key( + target_template.id, + candidate.unit.id, + candidate.function.id, + source_postbox.address_record.context_key, + ) + target_postbox = self._postbox_for_address_key( + session, + tenant_id=tenant_id, + address_key=key, + ) + if ( + candidate.status == "vacant" + and not target_revision.allow_vacant_delivery + ): + status = "vacancy_blocked" + diagnostics.append("target_template_blocks_vacancy") + return { + "depth": candidate.depth, + "organization_unit_id": candidate.unit.id, + "organization_unit_name": candidate.unit.name, + "function_id": ( + candidate.function.id if candidate.function else None + ), + "function_name": ( + candidate.function.name if candidate.function else None + ), + "target_postbox_id": ( + target_postbox.id if target_postbox else None + ), + "target_address": ( + target_postbox.address_record.address + if target_postbox + else None + ), + "status": status, + "vacant": candidate.holder_count == 0, + "holder_count": candidate.holder_count, + "path": [dict(item) for item in candidate.path], + "diagnostics": diagnostics, + } + + def _apply_hierarchy_routing( + self, + session: Session, + *, + request: PostboxDeliveryRequest, + delivery: PostboxDelivery, + source_postbox: Postbox, + source_message: PostboxMessage, + source_revision: PostboxTemplateRevision | None, + ) -> tuple[PostboxRoute, ...]: + plan = self._hierarchy_route_plan( + source_postbox=source_postbox, + source_revision=source_revision, + producer_module=request.producer_module, + classification=request.classification, + expires_at=request.expires_at, + ) + target_template, target_revision, template_diagnostics = ( + self._route_target_template( + session, + tenant_id=request.tenant_id, + plan=plan, + required=plan.status == "planned", + ) + ) + routing_snapshot: dict[str, object] = { + "status": ( + "blocked" + if template_diagnostics and plan.status == "planned" + else plan.status + ), + "policy": dict(plan.policy), + "diagnostics": list( + dict.fromkeys((*plan.diagnostics, *template_diagnostics)) + ), + "routes": [], + } + delivery.target_snapshot = { + **dict(delivery.target_snapshot or {}), + "hierarchy_routing": routing_snapshot, + } + if ( + plan.status != "planned" + or target_template is None + or target_revision is None + or template_diagnostics + ): + return () + + materialized: list[ + tuple[HierarchyRouteCandidate, Postbox] + ] = [] + seen_postbox_ids = {source_postbox.id} + for candidate in plan.candidates: + if ( + candidate.function is None + or candidate.status not in {"available", "vacant"} + ): + continue + if ( + candidate.status == "vacant" + and not target_revision.allow_vacant_delivery + ): + continue + try: + target_postbox = self.materialize_template( + session, + tenant_id=request.tenant_id, + template_id=target_template.id, + organization_unit_id=candidate.unit.id, + function_id=candidate.function.id, + context_key=source_postbox.address_record.context_key, + actor_id=None, + ) + except PostboxError as exc: + routing_snapshot["diagnostics"] = list( + dict.fromkeys( + ( + *routing_snapshot["diagnostics"], + f"target_{candidate.unit.id}:{exc.code}", + ) + ) + ) + continue + if ( + target_postbox.id in seen_postbox_ids + or target_postbox.status != "active" + ): + continue + seen_postbox_ids.add(target_postbox.id) + materialized.append((candidate, target_postbox)) + + linked_copy = _mapping(plan.policy.get("linked_copy")) + fanout = linked_copy.get("fanout", "nearest") + selected = materialized if fanout == "all" else materialized[:1] + routes: list[PostboxRoute] = [] + for candidate, target_postbox in selected: + route = self._create_linked_copy_route( + session, + request=request, + delivery=delivery, + source_postbox=source_postbox, + source_message=source_message, + target_postbox=target_postbox, + candidate=candidate, + policy=plan.policy, + route_kind="linked_copy", + source_route_id=None, + ) + routes.append(route) + + attention = _mapping(plan.policy.get("attention")) + if ( + fanout == "nearest" + and routes + and routes[0].status == "accepted_vacant" + and attention.get("mode") == "vacancy_escalation" + and len(materialized) > 1 + ): + remaining = [ + self._materialized_route_snapshot(candidate, postbox) + for candidate, postbox in materialized[1:] + ] + pending = self._schedule_escalation_route( + session, + delivery=delivery, + source_postbox=source_postbox, + source_message=source_message, + source_route=routes[0], + target=remaining[0], + remaining=remaining[1:], + policy=plan.policy, + now=utc_now(), + ) + routes.append(pending) + + routing_snapshot["routes"] = [ + self._route_evidence(route) for route in routes + ] + routing_snapshot["status"] = "routed" if routes else "no_route" + delivery.target_snapshot = { + **dict(delivery.target_snapshot or {}), + "hierarchy_routing": routing_snapshot, + } + source_message.metadata_ = { + **dict(source_message.metadata_ or {}), + "hierarchy_route_ids": [route.id for route in routes], + } + return tuple(routes) + + def _create_linked_copy_route( + self, + session: Session, + *, + request: PostboxDeliveryRequest, + delivery: PostboxDelivery, + source_postbox: Postbox, + source_message: PostboxMessage, + target_postbox: Postbox, + candidate: HierarchyRouteCandidate, + policy: Mapping[str, object], + route_kind: str, + source_route_id: str | None, + ) -> PostboxRoute: + now = utc_now() + route = PostboxRoute( + tenant_id=request.tenant_id, + delivery_id=delivery.id, + source_postbox_id=source_postbox.id, + source_message_id=source_message.id, + target_postbox_id=target_postbox.id, + target_message_id=None, + route_kind=route_kind, + status="processing", + depth=candidate.depth, + source_route_id=source_route_id, + execute_after=None, + processed_at=None, + policy_snapshot={ + "policy": dict(policy), + "target": self._materialized_route_snapshot( + candidate, + target_postbox, + ), + "evaluated_at": now.isoformat(), + "classification": request.classification, + "producer_module": request.producer_module, + }, + ) + session.add(route) + session.flush() + message = self._copy_routed_message( + session, + source_message=source_message, + target_postbox=target_postbox, + route=route, + delivered_at=now, + ) + route.target_message_id = message.id + route.processed_at = now + route.status = ( + "accepted" if candidate.holders else "accepted_vacant" + ) + session.flush() + self._record_access_event( + session, + tenant_id=request.tenant_id, + postbox_id=target_postbox.id, + message_id=message.id, + actor=None, + action=f"route.{route_kind}", + outcome="allowed", + reason_code=route.status, + details={ + "route_id": route.id, + "delivery_id": delivery.id, + "source_postbox_id": source_postbox.id, + "source_message_id": source_message.id, + "depth": candidate.depth, + }, + ) + _publish_postbox_event( + session, + f"postbox.route.{route_kind}.accepted.v1", + tenant_id=request.tenant_id, + resource_type="postbox_route", + resource_id=route.id, + postbox_id=target_postbox.id, + actor_type="module", + actor_id=request.producer_module, + payload=self._route_evidence(route), + ) + self._notify_delivery_holders( + session, + request=request, + delivery=delivery, + message=message, + holders=candidate.holders, + ) + return route + + def _copy_routed_message( + self, + session: Session, + *, + source_message: PostboxMessage, + target_postbox: Postbox, + route: PostboxRoute, + delivered_at: datetime, + ) -> PostboxMessage: + message = PostboxMessage( + tenant_id=source_message.tenant_id, + postbox_id=target_postbox.id, + subject=source_message.subject, + body_text=source_message.body_text, + status="delivered", + classification=source_message.classification, + sender_label=source_message.sender_label, + producer_module=source_message.producer_module, + producer_resource_type=source_message.producer_resource_type, + producer_resource_id=source_message.producer_resource_id, + encryption_profile=target_postbox.encryption_profile, + key_epoch=target_postbox.key_epoch, + delivered_at=delivered_at, + expires_at=source_message.expires_at, + retention_hold_until=source_message.retention_hold_until, + metadata_={ + **dict(source_message.metadata_ or {}), + "postbox_route": { + "route_id": route.id, + "route_kind": route.route_kind, + "source_postbox_id": source_message.postbox_id, + "source_message_id": source_message.id, + "delivery_id": route.delivery_id, + "depth": route.depth, + }, + }, + ) + for participant in source_message.participants: + message.participants.append( + PostboxParticipant( + tenant_id=source_message.tenant_id, + kind=participant.kind, + reference_type=participant.reference_type, + reference_id=participant.reference_id, + label=participant.label, + address=participant.address, + position=participant.position, + metadata_=dict(participant.metadata_ or {}), + ) + ) + for attachment in source_message.attachments: + message.attachments.append( + PostboxAttachmentReference( + tenant_id=source_message.tenant_id, + reference_type=attachment.reference_type, + reference_id=attachment.reference_id, + name=attachment.name, + media_type=attachment.media_type, + size_bytes=attachment.size_bytes, + digest=attachment.digest, + ciphertext_ref=attachment.ciphertext_ref, + position=attachment.position, + metadata_=dict(attachment.metadata_ or {}), + ) + ) + session.add(message) + session.flush() + return message + + def _schedule_escalation_route( + self, + session: Session, + *, + delivery: PostboxDelivery, + source_postbox: Postbox, + source_message: PostboxMessage, + source_route: PostboxRoute, + target: Mapping[str, object], + remaining: Sequence[Mapping[str, object]], + policy: Mapping[str, object], + now: datetime, + ) -> PostboxRoute: + attention = _mapping(policy.get("attention")) + delay_minutes = int(attention.get("delay_minutes") or 0) + route = PostboxRoute( + tenant_id=delivery.tenant_id, + delivery_id=delivery.id, + source_postbox_id=source_postbox.id, + source_message_id=source_message.id, + target_postbox_id=str(target["target_postbox_id"]), + target_message_id=None, + route_kind="attention_escalation", + status="pending_vacancy_escalation", + depth=int(target["depth"]), + source_route_id=source_route.id, + execute_after=now + timedelta(minutes=delay_minutes), + processed_at=None, + policy_snapshot={ + "policy": dict(policy), + "target": dict(target), + "remaining_targets": [dict(item) for item in remaining], + "evaluated_at": now.isoformat(), + "classification": source_message.classification, + "producer_module": delivery.producer_module, + }, + ) + session.add(route) + session.flush() + _publish_postbox_event( + session, + "postbox.route.attention_escalation.scheduled.v1", + tenant_id=delivery.tenant_id, + resource_type="postbox_route", + resource_id=route.id, + postbox_id=route.target_postbox_id, + actor_type="module", + actor_id=delivery.producer_module, + payload=self._route_evidence(route), + ) + return route + + def _dispatch_pending_route( + self, + session: Session, + *, + route: PostboxRoute, + now: datetime, + ) -> tuple[str, bool]: + if route.source_route_id: + previous_route = ( + session.query(PostboxRoute) + .filter( + PostboxRoute.id == route.source_route_id, + PostboxRoute.tenant_id == route.tenant_id, + ) + .one_or_none() + ) + previous_target = ( + self._get_postbox( + session, + tenant_id=route.tenant_id, + postbox_id=previous_route.target_postbox_id or "", + required=False, + ) + if previous_route is not None + else None + ) + previous_holders = ( + self._holders( + route.tenant_id, + previous_target.address_record.function_id, + ) + if previous_target is not None + else () + ) + acknowledged = ( + session.query(PostboxMessageReceipt) + .filter( + PostboxMessageReceipt.tenant_id == route.tenant_id, + PostboxMessageReceipt.message_id + == previous_route.target_message_id, + PostboxMessageReceipt.acknowledged_at.is_not(None), + ) + .count() + if ( + previous_route is not None + and previous_route.target_message_id + ) + else 0 + ) + if previous_holders or acknowledged: + route.status = ( + "cancelled_acknowledged" + if acknowledged + else "cancelled_vacancy_resolved" + ) + route.processed_at = now + return "cancelled", False + + source_message = self._get_message( + session, + tenant_id=route.tenant_id, + message_id=route.source_message_id, + ) + delivery = ( + session.query(PostboxDelivery) + .filter( + PostboxDelivery.id == route.delivery_id, + PostboxDelivery.tenant_id == route.tenant_id, + ) + .one_or_none() + ) + target_postbox = self._get_postbox( + session, + tenant_id=route.tenant_id, + postbox_id=route.target_postbox_id or "", + required=False, + ) + if ( + delivery is None + or target_postbox is None + or target_postbox.status != "active" + or source_message.withdrawn_at is not None + or source_message.expires_at is not None + and _as_utc(source_message.expires_at) <= _as_utc(now) + ): + route.status = "cancelled" + route.processed_at = now + return "cancelled", False + + target_revision = self._address_revision( + session, + target_postbox.address_record, + ) + holders = self._holders( + route.tenant_id, + target_postbox.address_record.function_id, + ) + if ( + not holders + and target_revision is not None + and not target_revision.allow_vacant_delivery + ): + route.status = "vacancy_blocked" + route.processed_at = now + return ( + "cancelled", + self._schedule_following_escalation( + session, + route=route, + delivery=delivery, + source_message=source_message, + now=now, + ), + ) + + message = self._copy_routed_message( + session, + source_message=source_message, + target_postbox=target_postbox, + route=route, + delivered_at=now, + ) + route.target_message_id = message.id + route.status = "accepted" if holders else "accepted_vacant" + route.processed_at = now + request = PostboxDeliveryRequest( + tenant_id=route.tenant_id, + target=PostboxTargetRef(postbox_id=target_postbox.id), + producer_module=delivery.producer_module, + producer_resource_type=delivery.producer_resource_type, + producer_resource_id=delivery.producer_resource_id, + idempotency_key=delivery.idempotency_key, + subject=source_message.subject, + body_text=source_message.body_text, + sender_label=source_message.sender_label, + classification=source_message.classification, + expires_at=source_message.expires_at, + metadata=dict(source_message.metadata_ or {}), + ) + self._notify_delivery_holders( + session, + request=request, + delivery=delivery, + message=message, + holders=holders, + ) + _publish_postbox_event( + session, + "postbox.route.attention_escalation.accepted.v1", + tenant_id=route.tenant_id, + resource_type="postbox_route", + resource_id=route.id, + postbox_id=target_postbox.id, + actor_type="module", + actor_id=delivery.producer_module, + payload=self._route_evidence(route), + ) + rescheduled = False + if not holders: + rescheduled = self._schedule_following_escalation( + session, + route=route, + delivery=delivery, + source_message=source_message, + now=now, + ) + return ("delivered" if holders else "vacant"), rescheduled + + def _schedule_following_escalation( + self, + session: Session, + *, + route: PostboxRoute, + delivery: PostboxDelivery, + source_message: PostboxMessage, + now: datetime, + ) -> bool: + snapshot = dict(route.policy_snapshot or {}) + remaining = [ + dict(item) + for item in snapshot.get("remaining_targets", []) + if isinstance(item, Mapping) + ] + if not remaining: + return False + source_postbox = self._get_postbox( + session, + tenant_id=route.tenant_id, + postbox_id=route.source_postbox_id, + ) + self._schedule_escalation_route( + session, + delivery=delivery, + source_postbox=source_postbox, + source_message=source_message, + source_route=route, + target=remaining[0], + remaining=remaining[1:], + policy=_mapping(snapshot.get("policy")), + now=now, + ) + return True + + def _materialized_route_snapshot( + self, + candidate: HierarchyRouteCandidate, + postbox: Postbox, + ) -> dict[str, object]: + return { + "depth": candidate.depth, + "organization_unit_id": candidate.unit.id, + "organization_unit_name": candidate.unit.name, + "organization_unit_type_id": candidate.unit.unit_type_id, + "function_id": ( + candidate.function.id if candidate.function else None + ), + "function_name": ( + candidate.function.name if candidate.function else None + ), + "function_type_id": ( + candidate.function.function_type_id + if candidate.function + else None + ), + "target_postbox_id": postbox.id, + "target_address": postbox.address_record.address, + "vacant_at_evaluation": candidate.holder_count == 0, + "holder_count_at_evaluation": candidate.holder_count, + "holder_assignment_ids": [ + holder.id for holder in candidate.holders + ], + "holder_assignment_sources": [ + holder.source for holder in candidate.holders + ], + "path": [dict(item) for item in candidate.path], + } + + def _route_evidence(self, route: PostboxRoute) -> dict[str, object]: + snapshot = dict(route.policy_snapshot or {}) + return { + "route_id": route.id, + "route_kind": route.route_kind, + "status": route.status, + "depth": route.depth, + "source_route_id": route.source_route_id, + "source_postbox_id": route.source_postbox_id, + "source_message_id": route.source_message_id, + "target_postbox_id": route.target_postbox_id, + "target_message_id": route.target_message_id, + "execute_after": ( + route.execute_after.isoformat() + if route.execute_after + else None + ), + "processed_at": ( + route.processed_at.isoformat() + if route.processed_at + else None + ), + "target_snapshot": snapshot.get("target", {}), + "policy_snapshot": snapshot.get("policy", {}), + } + def _notify_delivery_holders( self, session: Session, @@ -1138,6 +2075,7 @@ class PostboxService: classification: str, allow_vacant_delivery: bool, actor_id: str | None, + routing_policy: Mapping[str, object] | None = None, ) -> PostboxTemplate: clean_slug = _slug(slug or name, fallback="template") if ( @@ -1180,7 +2118,7 @@ class PostboxService: allow_vacant_delivery=allow_vacant_delivery, encryption_profile="plaintext_v1", history_policy={}, - routing_policy={"mode": "none"}, + routing_policy=normalized_routing_policy(routing_policy), retention_policy={}, created_by=actor_id, ) @@ -1213,6 +2151,7 @@ class PostboxService: classification: str, allow_vacant_delivery: bool, actor_id: str | None, + routing_policy: Mapping[str, object] | None = None, ) -> PostboxTemplate: template = self._get_template( session, @@ -1247,7 +2186,7 @@ class PostboxService: allow_vacant_delivery=allow_vacant_delivery, encryption_profile="plaintext_v1", history_policy={}, - routing_policy={"mode": "none"}, + routing_policy=normalized_routing_policy(routing_policy), retention_policy={}, created_by=actor_id, ) @@ -1504,6 +2443,49 @@ class PostboxService: ) return tuple(result) + def organization_hierarchy_targets( + self, + *, + tenant_id: str, + ) -> tuple[dict[str, object], ...]: + if self._hierarchy is None: + return () + catalog = self._hierarchy.hierarchy_catalog(tenant_id) + relation_types_by_structure: dict[ + str, + list[dict[str, object]], + ] = {} + for relation_type in catalog.relation_types: + if not relation_type.structure_id: + continue + relation_types_by_structure.setdefault( + relation_type.structure_id, + [], + ).append( + { + "id": relation_type.id, + "slug": relation_type.slug, + "name": relation_type.name, + "structure_id": relation_type.structure_id, + "is_hierarchical": relation_type.is_hierarchical, + "status": relation_type.status, + } + ) + return tuple( + { + "id": structure.id, + "slug": structure.slug, + "name": structure.name, + "structure_kind": structure.structure_kind, + "status": structure.status, + "relation_types": relation_types_by_structure.get( + structure.id, + [], + ), + } + for structure in catalog.structures + ) + # Grouping projections def list_groupings( self, @@ -2117,6 +3099,25 @@ class PostboxService: *, duplicate: bool = False, ) -> PostboxDeliveryResult: + session = object_session(delivery) + route_evidence = ( + [ + self._route_evidence(route) + for route in session.query(PostboxRoute) + .filter( + PostboxRoute.tenant_id == delivery.tenant_id, + PostboxRoute.delivery_id == delivery.id, + ) + .order_by( + PostboxRoute.depth, + PostboxRoute.created_at, + PostboxRoute.id, + ) + .all() + ] + if session is not None + else [] + ) return PostboxDeliveryResult( delivery_id=delivery.id, postbox_id=delivery.postbox_id, @@ -2133,6 +3134,7 @@ class PostboxService: "template_revision_id": delivery.template_revision_id, "target_snapshot": dict(delivery.target_snapshot or {}), "accepted_at": delivery.accepted_at.isoformat(), + "hierarchy_routes": route_evidence, }, ) diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 311a23c..f7db00f 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -8,6 +8,7 @@ from govoplan_core.core.postbox import ( CAPABILITY_POSTBOX_DIRECTORY, CAPABILITY_POSTBOX_EVIDENCE, CAPABILITY_POSTBOX_MESSAGES, + CAPABILITY_POSTBOX_ROUTING, ) from govoplan_postbox.backend.manifest import get_manifest @@ -28,6 +29,7 @@ class PostboxManifestTests(unittest.TestCase): CAPABILITY_POSTBOX_MESSAGES, CAPABILITY_POSTBOX_DELIVERY, CAPABILITY_POSTBOX_EVIDENCE, + CAPABILITY_POSTBOX_ROUTING, }, set(manifest.capability_factories), ) diff --git a/tests/test_migration.py b/tests/test_migration.py index 0afeedb..f1ae553 100644 --- a/tests/test_migration.py +++ b/tests/test_migration.py @@ -14,14 +14,21 @@ class PostboxMigrationTests(unittest.TestCase): "govoplan_postbox.backend.migrations.versions." "c7d2e5f8a1b4_v010_postbox_baseline" ) + route_migration = importlib.import_module( + "govoplan_postbox.backend.migrations.versions." + "e4b7c9d2a6f1_v011_hierarchy_routes" + ) engine = create_engine("sqlite:///:memory:") try: with engine.begin() as connection: operations = Operations(MigrationContext.configure(connection)) original = migration.op + route_original = route_migration.op migration.op = operations + route_migration.op = operations try: migration.upgrade() + route_migration.upgrade() tables = set(inspect(connection).get_table_names()) self.assertIn("postboxes", tables) self.assertIn("postbox_messages", tables) @@ -43,6 +50,18 @@ class PostboxMigrationTests(unittest.TestCase): "withdrawn_at", }.issubset(message_columns) ) + route_columns = { + column["name"] + for column in inspect(connection).get_columns( + "postbox_routes" + ) + } + self.assertTrue( + {"execute_after", "processed_at"}.issubset( + route_columns + ) + ) + route_migration.downgrade() migration.downgrade() self.assertFalse( { @@ -53,6 +72,7 @@ class PostboxMigrationTests(unittest.TestCase): ) finally: migration.op = original + route_migration.op = route_original finally: engine.dispose() diff --git a/tests/test_router.py b/tests/test_router.py index 01e45bc..c96379e 100644 --- a/tests/test_router.py +++ b/tests/test_router.py @@ -297,6 +297,23 @@ class PostboxRouterTests(unittest.TestCase): self.assertEqual(200, unread.status_code, unread.text) self.assertEqual(0, unread.json()["total"]) + def test_routing_dry_run_explains_default_disabled_state(self) -> None: + response = self.client.post( + "/api/v1/postbox/routing/dry-run", + json={ + "target": {"postbox_id": self.postbox_id}, + "producer_module": "campaigns", + "classification": "internal", + }, + ) + + self.assertEqual(200, response.status_code, response.text) + self.assertEqual("disabled", response.json()["status"]) + self.assertEqual( + ["hierarchy_routing_disabled"], + response.json()["diagnostics"], + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_service.py b/tests/test_service.py index 6d72976..7b66500 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -14,7 +14,17 @@ from govoplan_core.core.idm import ( ) from govoplan_core.core.organizations import ( OrganizationFunctionRef, + OrganizationFunctionTypeRef, + OrganizationFunctionTypeResolution, + OrganizationHierarchyCatalogRef, + OrganizationHierarchyEdgeRef, + OrganizationHierarchyMatchRef, + OrganizationHierarchyPathResolution, + OrganizationHierarchyResolution, + OrganizationRelationTypeRef, + OrganizationStructureRef, OrganizationUnitRef, + OrganizationUnitTypeResolution, ) from govoplan_core.core.postbox import ( PostboxActorRef, @@ -173,6 +183,13 @@ class FakeOrganizationDirectory: unit_type_id="desk", parent_id="unit-1", ), + "unit-top": OrganizationUnitRef( + id="unit-top", + tenant_id="tenant-1", + slug="central-office", + name="Central Office", + unit_type_id="office", + ), } self.functions = { "function-1": OrganizationFunctionRef( @@ -193,7 +210,47 @@ class FakeOrganizationDirectory: function_type_id="case-clerk-type", delegable=True, ), + "function-top": OrganizationFunctionRef( + id="function-top", + tenant_id="tenant-1", + organization_unit_id="unit-top", + slug="case-clerk", + name="Case Clerk", + function_type_id="case-clerk-type", + delegable=True, + ), } + self.structure = OrganizationStructureRef( + id="structure-1", + tenant_id="tenant-1", + slug="administrative", + name="Administrative hierarchy", + structure_kind="administrative", + ) + self.parallel_structure = OrganizationStructureRef( + id="structure-2", + tenant_id="tenant-1", + slug="reporting", + name="Reporting hierarchy", + structure_kind="reporting", + ) + self.relation_type = OrganizationRelationTypeRef( + id="relation-type-1", + tenant_id="tenant-1", + slug="reports-to", + name="Reports to", + structure_id=self.structure.id, + ) + self.parallel_relation_type = OrganizationRelationTypeRef( + id="relation-type-2", + tenant_id="tenant-1", + slug="reports-to", + name="Reports to", + structure_id=self.parallel_structure.id, + ) + self.last_hierarchy_request: dict[str, object] = {} + self.duplicate_parent_match = False + self.cycle_detected = False def get_organization_unit(self, organization_unit_id: str): return self.units.get(organization_unit_id) @@ -218,6 +275,214 @@ class FakeOrganizationDirectory: if function.organization_unit_id == organization_unit_id ) + def hierarchy_catalog(self, tenant_id: str): + if tenant_id != "tenant-1": + return OrganizationHierarchyCatalogRef(tenant_id=tenant_id) + return OrganizationHierarchyCatalogRef( + tenant_id=tenant_id, + structures=(self.structure, self.parallel_structure), + relation_types=( + self.relation_type, + self.parallel_relation_type, + ), + ) + + def get_unit_type(self, tenant_id: str, unit_type_id: str): + del tenant_id, unit_type_id + return None + + def get_function_type(self, tenant_id: str, function_type_id: str): + if tenant_id != "tenant-1" or function_type_id != "case-clerk-type": + return None + return OrganizationFunctionTypeRef( + id=function_type_id, + tenant_id=tenant_id, + slug="case-clerk", + name="Case Clerk", + ) + + def resolve_functions_by_type( + self, + tenant_id: str, + function_type_id: str, + *, + organization_unit_ids=(), + ): + if tenant_id != "tenant-1": + return OrganizationFunctionTypeResolution( + tenant_id=tenant_id, + function_type_id=function_type_id, + requested_unit_ids=tuple(organization_unit_ids), + status="missing", + ) + return OrganizationFunctionTypeResolution( + tenant_id=tenant_id, + function_type_id=function_type_id, + requested_unit_ids=tuple(organization_unit_ids), + status="active", + function_type=self.get_function_type( + tenant_id, + function_type_id, + ), + matches=tuple( + function + for function in self.functions.values() + if function.function_type_id == function_type_id + and ( + not organization_unit_ids + or function.organization_unit_id + in organization_unit_ids + ) + ), + ) + + def resolve_units_by_type( + self, + tenant_id: str, + unit_type_id: str, + **kwargs, + ): + del kwargs + return OrganizationUnitTypeResolution( + tenant_id=tenant_id, + unit_type_id=unit_type_id, + status="active", + matches=tuple( + unit + for unit in self.units.values() + if unit.tenant_id == tenant_id + and unit.unit_type_id == unit_type_id + ), + ) + + def resolve_hierarchy_relatives( + self, + tenant_id: str, + organization_unit_ids, + *, + structure_id: str, + relation_type_ids=(), + direction="ancestors", + max_depth=10, + ): + self.last_hierarchy_request = { + "tenant_id": tenant_id, + "structure_id": structure_id, + "relation_type_ids": tuple(relation_type_ids), + "direction": direction, + "max_depth": max_depth, + } + results = [] + for root_id in organization_unit_ids: + root = self.units.get(root_id) + if root is None or root.tenant_id != tenant_id: + results.append( + OrganizationHierarchyResolution( + tenant_id=tenant_id, + root_unit_id=root_id, + direction=direction, + structure_id=structure_id, + relation_type_ids=tuple(relation_type_ids), + max_depth=max_depth, + status="missing", + ) + ) + continue + if structure_id == self.parallel_structure.id: + matches = () + else: + first_edge = OrganizationHierarchyEdgeRef( + id="edge-child-parent", + tenant_id=tenant_id, + structure=self.structure, + relation_type=self.relation_type, + source_unit_id=root_id, + target_unit_id="unit-1", + ) + second_edge = OrganizationHierarchyEdgeRef( + id="edge-parent-top", + tenant_id=tenant_id, + structure=self.structure, + relation_type=self.relation_type, + source_unit_id="unit-1", + target_unit_id="unit-top", + ) + available = ( + OrganizationHierarchyMatchRef( + unit=self.units["unit-1"], + depth=1, + path=(first_edge,), + ), + OrganizationHierarchyMatchRef( + unit=self.units["unit-top"], + depth=2, + path=(first_edge, second_edge), + ), + ) + if self.duplicate_parent_match: + duplicate_edge = OrganizationHierarchyEdgeRef( + id="edge-child-parent-duplicate", + tenant_id=tenant_id, + structure=self.structure, + relation_type=self.relation_type, + source_unit_id=root_id, + target_unit_id="unit-1", + ) + available = ( + available[0], + OrganizationHierarchyMatchRef( + unit=self.units["unit-1"], + depth=1, + path=(duplicate_edge,), + ), + available[1], + ) + matches = tuple( + match for match in available if match.depth <= max_depth + ) + results.append( + OrganizationHierarchyResolution( + tenant_id=tenant_id, + root_unit_id=root_id, + direction=direction, + structure_id=structure_id, + relation_type_ids=tuple(relation_type_ids), + max_depth=max_depth, + status="active", + root=root, + matches=matches, + cycle_detected=self.cycle_detected, + depth_limited=( + structure_id == self.structure.id and max_depth < 2 + ), + ) + ) + return tuple(results) + + def resolve_hierarchy_paths( + self, + tenant_id: str, + unit_pairs, + *, + structure_id: str, + relation_type_ids=(), + direction="descendants", + max_depth=10, + ): + return tuple( + OrganizationHierarchyPathResolution( + tenant_id=tenant_id, + source_unit_id=source_id, + target_unit_id=target_id, + direction=direction, + structure_id=structure_id, + relation_type_ids=tuple(relation_type_ids), + max_depth=max_depth, + status="unreachable", + ) + for source_id, target_id in unit_pairs + ) + class FakeNotificationDispatch: def __init__(self) -> None: @@ -280,6 +545,117 @@ class PostboxServiceTests(unittest.TestCase): actor_id="admin-1", ) + def _routing_policy( + self, + target_template_id: str, + *, + structure_id: str = "structure-1", + max_depth: int = 2, + producer_modules: tuple[str, ...] = ("campaigns",), + classifications: tuple[str, ...] = ("internal",), + vacancy_escalation: bool = False, + ) -> dict[str, object]: + return { + "linked_copy": { + "enabled": True, + "structure_id": structure_id, + "relation_type_ids": ["relation-type-1"], + "max_depth": max_depth, + "target_function_type_id": "case-clerk-type", + "target_template_id": target_template_id, + "fanout": "nearest", + "allowed_classifications": list(classifications), + "allowed_producer_modules": list(producer_modules), + "require_expiry": True, + "max_retention_days": 30, + }, + "attention": { + "mode": ( + "vacancy_escalation" + if vacancy_escalation + else "none" + ), + "delay_minutes": 1 if vacancy_escalation else None, + }, + "shared_visibility": {"mode": "none"}, + } + + def _create_routing_source( + self, + session: Session, + *, + vacancy_escalation: bool = False, + structure_id: str = "structure-1", + max_depth: int = 2, + ) -> tuple[PostboxService, Postbox, PostboxTemplate]: + service = PostboxService( + identities=FakeIdentityDirectory(), # type: ignore[arg-type] + idm=self.idm, # type: ignore[arg-type] + incumbencies=self.idm, # type: ignore[arg-type] + organizations=self.organizations, # type: ignore[arg-type] + hierarchy=self.organizations, # type: ignore[arg-type] + ) + target_template = service.create_template( + session, + tenant_id="tenant-1", + slug="hierarchy-target", + name="Hierarchy target", + description=None, + function_type_id="case-clerk-type", + scope_kind="tenant", + scope_id=None, + name_pattern="{unit_name} / {function_name} Routed", + address_pattern="routed.{unit_slug}.{function_slug}", + classification="internal", + allow_vacant_delivery=True, + actor_id="admin-1", + ) + service.publish_template( + session, + tenant_id="tenant-1", + template_id=target_template.id, + revision_number=None, + actor_id="admin-1", + ) + source_template = service.create_template( + session, + tenant_id="tenant-1", + slug="hierarchy-source", + name="Hierarchy source", + description=None, + function_type_id="case-clerk-type", + scope_kind="tenant", + scope_id=None, + name_pattern="{unit_name} / {function_name} Source", + address_pattern="source.{unit_slug}.{function_slug}", + classification="internal", + allow_vacant_delivery=True, + actor_id="admin-1", + routing_policy=self._routing_policy( + target_template.id, + vacancy_escalation=vacancy_escalation, + structure_id=structure_id, + max_depth=max_depth, + ), + ) + service.publish_template( + session, + tenant_id="tenant-1", + template_id=source_template.id, + revision_number=None, + actor_id="admin-1", + ) + source = service.materialize_template( + session, + tenant_id="tenant-1", + template_id=source_template.id, + organization_unit_id="unit-child", + function_id="function-child", + context_key="case-42", + actor_id="admin-1", + ) + return service, source, target_template + def test_access_follows_current_assignment_and_reports_vacancy(self) -> None: with Session(self.engine) as session: postbox = self._create_exact(session) @@ -595,6 +971,260 @@ class PostboxServiceTests(unittest.TestCase): [event.type for event in events], ) + def test_hierarchy_linked_copy_snapshots_path_and_independent_state( + self, + ) -> None: + self.idm.assignments.append(self.assignment) + with Session(self.engine) as session: + service, source, _target_template = ( + self._create_routing_source(session) + ) + result = service.deliver( + session, + PostboxDeliveryRequest( + tenant_id="tenant-1", + target=PostboxTargetRef(postbox_id=source.id), + producer_module="campaigns", + producer_resource_type="campaign_recipient", + producer_resource_id="recipient-1", + idempotency_key="routing-copy-1", + subject="Routed decision", + body_text="The decision is available.", + classification="internal", + expires_at=utc_now() + timedelta(days=7), + ), + ) + session.commit() + + routes = session.query(PostboxRoute).all() + self.assertEqual(1, len(routes)) + route = routes[0] + self.assertEqual("linked_copy", route.route_kind) + self.assertEqual("accepted", route.status) + self.assertNotEqual(result.message_id, route.target_message_id) + self.assertEqual( + ["edge-child-parent"], + [ + edge["edge_id"] + for edge in route.policy_snapshot["target"]["path"] + ], + ) + self.assertEqual( + route.id, + result.evidence["hierarchy_routes"][0]["route_id"], + ) + service.mark_message( + session, + tenant_id="tenant-1", + message_id=route.target_message_id, + actor=self.actor, + state="read", + ) + session.commit() + receipts = session.query(PostboxMessageReceipt).all() + self.assertEqual([route.target_message_id], [item.message_id for item in receipts]) + + def test_hierarchy_dry_run_explains_gates_depth_and_parallel_structure( + self, + ) -> None: + with Session(self.engine) as session: + service, source, _target_template = ( + self._create_routing_source(session, max_depth=1) + ) + blocked = service.preview_hierarchy_routes( + session, + tenant_id="tenant-1", + target=PostboxTargetRef(postbox_id=source.id), + producer_module="mail", + classification="restricted", + expires_at=None, + ) + self.assertEqual("blocked", blocked["status"]) + self.assertTrue( + { + "classification_not_allowed", + "producer_not_authorized", + "expiry_required", + }.issubset(set(blocked["diagnostics"])) + ) + + planned = service.preview_hierarchy_routes( + session, + tenant_id="tenant-1", + target=PostboxTargetRef(postbox_id=source.id), + producer_module="campaigns", + classification="internal", + expires_at=utc_now() + timedelta(days=7), + ) + self.assertEqual("planned", planned["status"]) + self.assertEqual(1, len(planned["routes"])) + self.assertEqual( + 1, + self.organizations.last_hierarchy_request["max_depth"], + ) + self.assertIn( + "hierarchy_depth_limited", + planned["diagnostics"], + ) + + with Session(self.engine) as session: + service, source, _target_template = self._create_routing_source( + session, + structure_id="structure-2", + ) + no_route = service.preview_hierarchy_routes( + session, + tenant_id="tenant-1", + target=PostboxTargetRef(postbox_id=source.id), + producer_module="campaigns", + classification="internal", + expires_at=utc_now() + timedelta(days=7), + ) + self.assertEqual("no_route", no_route["status"]) + self.assertIn("no_hierarchy_ancestor", no_route["diagnostics"]) + + def test_hierarchy_preview_bounds_cycles_deduplicates_and_is_tenant_safe( + self, + ) -> None: + self.organizations.duplicate_parent_match = True + self.organizations.cycle_detected = True + with Session(self.engine) as session: + service, source, _target_template = ( + self._create_routing_source(session) + ) + preview = service.preview_hierarchy_routes( + session, + tenant_id="tenant-1", + target=PostboxTargetRef(postbox_id=source.id), + producer_module="campaigns", + classification="internal", + expires_at=utc_now() + timedelta(days=7), + ) + + self.assertIn("hierarchy_cycle_bounded", preview["diagnostics"]) + self.assertEqual( + 1, + sum( + route["status"] == "duplicate" + for route in preview["routes"] + ), + ) + with self.assertRaisesRegex( + ValueError, + "existing source Postbox", + ): + service.preview_hierarchy_routes( + session, + tenant_id="tenant-2", + target=PostboxTargetRef(postbox_id=source.id), + producer_module="campaigns", + classification="internal", + expires_at=utc_now() + timedelta(days=7), + ) + + def test_vacancy_escalation_is_delayed_and_uses_frozen_targets( + self, + ) -> None: + top_assignment = OrganizationFunctionAssignmentRef( + id="assignment-top", + tenant_id="tenant-1", + identity_id="identity-top", + account_id="account-top", + function_id="function-top", + organization_unit_id="unit-top", + source="direct", + ) + self.idm.assignments.append(top_assignment) + with Session(self.engine) as session: + service, source, _target_template = self._create_routing_source( + session, + vacancy_escalation=True, + ) + service.deliver( + session, + PostboxDeliveryRequest( + tenant_id="tenant-1", + target=PostboxTargetRef(postbox_id=source.id), + producer_module="campaigns", + producer_resource_type="campaign_recipient", + producer_resource_id="recipient-2", + idempotency_key="routing-vacancy-1", + subject="Escalated decision", + classification="internal", + expires_at=utc_now() + timedelta(days=7), + ), + ) + session.commit() + routes = ( + session.query(PostboxRoute) + .order_by(PostboxRoute.depth) + .all() + ) + self.assertEqual( + ["accepted_vacant", "pending_vacancy_escalation"], + [route.status for route in routes], + ) + pending = routes[1] + frozen_target_id = pending.target_postbox_id + pending.execute_after = utc_now() - timedelta(seconds=1) + session.commit() + + result = service.dispatch_due_routes(session) + session.commit() + session.refresh(pending) + self.assertEqual(1, result["delivered"]) + self.assertEqual("accepted", pending.status) + self.assertEqual(frozen_target_id, pending.target_postbox_id) + self.assertIsNotNone(pending.target_message_id) + self.assertEqual(3, session.query(PostboxMessage).count()) + + def test_vacancy_escalation_stops_when_previous_function_is_filled( + self, + ) -> None: + with Session(self.engine) as session: + service, source, _target_template = self._create_routing_source( + session, + vacancy_escalation=True, + ) + service.deliver( + session, + PostboxDeliveryRequest( + tenant_id="tenant-1", + target=PostboxTargetRef(postbox_id=source.id), + producer_module="campaigns", + producer_resource_type="campaign_recipient", + producer_resource_id="recipient-3", + idempotency_key="routing-vacancy-resolved-1", + subject="No longer escalated", + classification="internal", + expires_at=utc_now() + timedelta(days=7), + ), + ) + session.commit() + pending = ( + session.query(PostboxRoute) + .filter( + PostboxRoute.status + == "pending_vacancy_escalation" + ) + .one() + ) + pending.execute_after = utc_now() - timedelta(seconds=1) + self.idm.assignments.append(self.assignment) + session.commit() + + result = service.dispatch_due_routes(session) + session.commit() + session.refresh(pending) + + self.assertEqual(1, result["cancelled"]) + self.assertEqual( + "cancelled_vacancy_resolved", + pending.status, + ) + self.assertIsNone(pending.target_message_id) + self.assertEqual(2, session.query(PostboxMessage).count()) + def test_grouping_update_retains_temporarily_hidden_sources(self) -> None: child_assignment = OrganizationFunctionAssignmentRef( id="assignment-child", diff --git a/webui/src/api/postbox.ts b/webui/src/api/postbox.ts index f17701d..9e47e3c 100644 --- a/webui/src/api/postbox.ts +++ b/webui/src/api/postbox.ts @@ -110,6 +110,54 @@ export type PostboxOrganizationUnit = { functions: PostboxOrganizationFunction[]; }; +export type PostboxOrganizationRelationType = { + id: string; + slug: string; + name: string; + structure_id?: string | null; + is_hierarchical: boolean; + status: string; +}; + +export type PostboxOrganizationStructure = { + id: string; + slug: string; + name: string; + structure_kind: string; + status: string; + relation_types: PostboxOrganizationRelationType[]; +}; + +export type PostboxOrganizationTargets = { + units: PostboxOrganizationUnit[]; + structures: PostboxOrganizationStructure[]; +}; + +export type PostboxRoutingPolicy = { + linked_copy: { + enabled: boolean; + structure_id?: string | null; + relation_type_ids: string[]; + max_depth: number; + stop_unit_id?: string | null; + stop_unit_type_id?: string | null; + target_function_type_id?: string | null; + target_template_id?: string | null; + fanout: "nearest" | "all"; + allowed_classifications: string[]; + allowed_producer_modules: string[]; + require_expiry: boolean; + max_retention_days?: number | null; + }; + attention: { + mode: "none" | "vacancy_escalation"; + delay_minutes?: number | null; + }; + shared_visibility: { + mode: "none"; + }; +}; + export type PostboxTemplateRevision = { id: string; revision: number; @@ -122,7 +170,7 @@ export type PostboxTemplateRevision = { allow_vacant_delivery: boolean; encryption_profile: string; history_policy: Record; - routing_policy: Record; + routing_policy: PostboxRoutingPolicy; retention_policy: Record; published_at?: string | null; created_at: string; @@ -151,6 +199,7 @@ export type PostboxTemplateRevisionPayload = Pick< | "address_pattern" | "classification" | "allow_vacant_delivery" + | "routing_policy" >; export type PostboxTemplateCreatePayload = PostboxTemplateRevisionPayload & { @@ -269,12 +318,11 @@ export async function listAdminPostboxes(settings: ApiSettings): Promise { - const response = await apiFetch<{ units: PostboxOrganizationUnit[] }>( +): Promise { + return apiFetch( settings, "/api/v1/postbox/admin/organization-targets" ); - return response.units; } export function createExactPostbox( diff --git a/webui/src/features/postbox/PostboxAdminPanel.tsx b/webui/src/features/postbox/PostboxAdminPanel.tsx index c475a5b..0555634 100644 --- a/webui/src/features/postbox/PostboxAdminPanel.tsx +++ b/webui/src/features/postbox/PostboxAdminPanel.tsx @@ -39,7 +39,9 @@ import { type PostboxDirectoryItem, type PostboxExactCreatePayload, type PostboxOrganizationFunction, + type PostboxOrganizationStructure, type PostboxOrganizationUnit, + type PostboxRoutingPolicy, type PostboxTemplate, type PostboxTemplateCreatePayload, type PostboxTemplateRevisionPayload @@ -56,6 +58,31 @@ type MaterializeDraft = { context_key: string; }; +const routingDefaults = (): PostboxRoutingPolicy => ({ + linked_copy: { + enabled: false, + structure_id: null, + relation_type_ids: [], + max_depth: 1, + stop_unit_id: null, + stop_unit_type_id: null, + target_function_type_id: null, + target_template_id: null, + fanout: "nearest", + allowed_classifications: ["internal"], + allowed_producer_modules: [], + require_expiry: false, + max_retention_days: null + }, + attention: { + mode: "none", + delay_minutes: null + }, + shared_visibility: { + mode: "none" + } +}); + const templateDefaults = (): TemplateDraft => ({ templateId: "", slug: "", @@ -67,7 +94,8 @@ const templateDefaults = (): TemplateDraft => ({ name_pattern: "{unit_name} / {function_name}", address_pattern: "{template_slug}.{unit_slug}.{function_slug}", classification: "internal", - allow_vacant_delivery: true + allow_vacant_delivery: true, + routing_policy: routingDefaults() }); const exactDefaults = (): ExactDraft => ({ @@ -94,6 +122,7 @@ export default function PostboxAdminPanel({ const [templates, setTemplates] = useState([]); const [postboxes, setPostboxes] = useState([]); const [units, setUnits] = useState([]); + const [structures, setStructures] = useState([]); const [selectedTemplateId, setSelectedTemplateId] = useState(""); const [selectedPostboxId, setSelectedPostboxId] = useState(""); const [loading, setLoading] = useState(true); @@ -146,14 +175,15 @@ export default function PostboxAdminPanel({ setLoading(true); setError(""); try { - const [nextTemplates, nextPostboxes, nextUnits] = await Promise.all([ + const [nextTemplates, nextPostboxes, organizationTargets] = await Promise.all([ canManageTemplates ? listPostboxTemplates(settings) : Promise.resolve([]), canManageBindings ? listAdminPostboxes(settings) : Promise.resolve([]), listPostboxOrganizationTargets(settings) ]); setTemplates(nextTemplates); setPostboxes(nextPostboxes); - setUnits(nextUnits); + setUnits(organizationTargets.units); + setStructures(organizationTargets.structures); setSelectedTemplateId((current) => current && nextTemplates.some((template) => template.id === current) ? current @@ -194,7 +224,8 @@ export default function PostboxAdminPanel({ name_pattern: revision.name_pattern, address_pattern: revision.address_pattern, classification: revision.classification, - allow_vacant_delivery: revision.allow_vacant_delivery + allow_vacant_delivery: revision.allow_vacant_delivery, + routing_policy: revision.routing_policy ?? routingDefaults() }); setTemplateDialogOpen(true); } @@ -409,6 +440,8 @@ export default function PostboxAdminPanel({ open={templateDialogOpen} draft={templateDraft} units={units} + structures={structures} + templates={templates} functionTypes={functionTypes} unitTypes={unitTypes} busy={busy} @@ -531,6 +564,22 @@ function TemplateWorkspace({
Scope
{revision.scope_kind}{revision.scope_id ? ` · ${revision.scope_id}` : ""}
Classification
{revision.classification}
Vacant delivery
{revision.allow_vacant_delivery ? "Accepted" : "Blocked"}
+
+
Hierarchy copies
+
+ {revision.routing_policy.linked_copy.enabled + ? `${revision.routing_policy.linked_copy.fanout} · depth ${revision.routing_policy.linked_copy.max_depth}` + : "Disabled"} +
+
+
+
Vacancy escalation
+
+ {revision.routing_policy.attention.mode === "vacancy_escalation" + ? `${revision.routing_policy.attention.delay_minutes} minutes` + : "Disabled"} +
+
Encryption
{revision.encryption_profile}
Name pattern
{revision.name_pattern}
Address pattern
{revision.address_pattern}
@@ -642,6 +691,8 @@ function TemplateDialog({ open, draft, units, + structures, + templates, functionTypes, unitTypes, busy, @@ -652,6 +703,8 @@ function TemplateDialog({ open: boolean; draft: TemplateDraft; units: PostboxOrganizationUnit[]; + structures: PostboxOrganizationStructure[]; + templates: PostboxTemplate[]; functionTypes: Array<{ id: string; name: string }>; unitTypes: Array<{ id: string; example: string }>; busy: boolean; @@ -663,12 +716,41 @@ function TemplateDialog({ const scopeOptions = draft.scope_kind === "unit_type" ? unitTypes.map((item) => ({ id: item.id, label: `${item.id} (${item.example})` })) : units.map((unit) => ({ id: unit.id, label: unit.name })); + const linkedCopy = draft.routing_policy.linked_copy; + const attention = draft.routing_policy.attention; + const selectedStructure = structures.find( + (item) => item.id === linkedCopy.structure_id + ); + const updateRouting = (routing_policy: PostboxRoutingPolicy) => { + onChange({ ...draft, routing_policy }); + }; + const updateLinkedCopy = ( + next: Partial + ) => { + updateRouting({ + ...draft.routing_policy, + linked_copy: { + ...linkedCopy, + ...next + } + }); + }; const valid = draft.name.trim() && draft.slug.trim() && draft.name_pattern.trim() && draft.address_pattern.trim() && - (draft.scope_kind === "tenant" || Boolean(draft.scope_id)); + (draft.scope_kind === "tenant" || Boolean(draft.scope_id)) && + ( + !linkedCopy.enabled + || Boolean( + linkedCopy.structure_id + && linkedCopy.target_function_type_id + && linkedCopy.target_template_id + && linkedCopy.allowed_classifications.length + && linkedCopy.allowed_producer_modules.length + ) + ); return ( onChange({ ...draft, allow_vacant_delivery: checked })} /> +
+
+
+ Hierarchy linked copies + Copy to explicitly bounded function Postboxes in one selected structure. +
+ updateLinkedCopy({ + enabled, + allowed_classifications: linkedCopy.allowed_classifications.length + ? linkedCopy.allowed_classifications + : [draft.classification] + })} + /> +
+ {linkedCopy.enabled ? ( +
+ + + + + + + + + + + + + + updateLinkedCopy({ + max_depth: Math.max(1, Math.min(20, Number(event.target.value) || 1)) + })} + /> + + + + + + + + + + + + updateLinkedCopy({ + allowed_classifications: commaSeparated(event.target.value) + })} + /> + + + updateLinkedCopy({ + allowed_producer_modules: commaSeparated(event.target.value) + })} + /> + +
+ updateLinkedCopy({ require_expiry })} + /> +
+ + updateLinkedCopy({ + max_retention_days: event.target.value + ? Math.max(1, Math.min(36500, Number(event.target.value))) + : null + })} + /> + +
+ updateRouting({ + ...draft.routing_policy, + attention: checked + ? { + mode: "vacancy_escalation", + delay_minutes: attention.delay_minutes || 1440 + } + : { mode: "none", delay_minutes: null } + })} + /> +
+ {attention.mode === "vacancy_escalation" ? ( + + updateRouting({ + ...draft.routing_policy, + attention: { + mode: "vacancy_escalation", + delay_minutes: Math.max( + 1, + Math.min(43200, Number(event.target.value) || 1) + ) + } + })} + /> + + ) :
} +
+ ) : null} +

Available pattern variables include template, unit, function, and optional context names or slugs. Published revisions are immutable. @@ -954,7 +1242,8 @@ function revisionPayload(draft: TemplateDraft): PostboxTemplateRevisionPayload { name_pattern: draft.name_pattern, address_pattern: draft.address_pattern, classification: draft.classification, - allow_vacant_delivery: draft.allow_vacant_delivery + allow_vacant_delivery: draft.allow_vacant_delivery, + routing_policy: draft.routing_policy }; } @@ -975,3 +1264,7 @@ function compatibleTargets( function errorMessage(error: unknown): string { return error instanceof Error ? error.message : "Postbox request failed"; } + +function commaSeparated(value: string): string[] { + return [...new Set(value.split(",").map((item) => item.trim()).filter(Boolean))]; +} diff --git a/webui/src/styles/postbox.css b/webui/src/styles/postbox.css index 5e7f209..455c69c 100644 --- a/webui/src/styles/postbox.css +++ b/webui/src/styles/postbox.css @@ -441,6 +441,31 @@ padding-bottom: 7px; } +.postbox-routing-section { + grid-column: 1 / -1; + display: grid; + gap: 14px; + border-top: var(--border-line); + padding-top: 16px; +} + +.postbox-routing-heading { + display: flex; + align-items: center; + justify-content: space-between; + gap: 18px; +} + +.postbox-routing-heading > div:first-child { + display: grid; + gap: 3px; +} + +.postbox-routing-heading span { + color: var(--muted); + font-size: 12px; +} + .postbox-form-note { margin: 15px 0 0; color: var(--muted); @@ -637,4 +662,9 @@ .postbox-message-detail { padding: 16px; } + + .postbox-routing-heading { + align-items: flex-start; + flex-direction: column; + } }