GovOPlaN/govoplan-postbox
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Add Postbox message authoring and concurrency

Browse Source
This commit is contained in:
zemion
2026-07-31 18:40:39 +02:00
parent 0b6ebc3c74
commit dc6f81dd33
12 changed files with 1392 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_postbox/backend/service.py
+414 -2
View File
@@ -1,12 +1,13 @@
from __future__ import annotations
import hashlib
import json
import logging
import re
from collections import Counter
from collections.abc import Mapping, Sequence
from datetime import datetime, timedelta, timezone
from typing import Literal
from typing import Any, Literal
from sqlalchemy import and_, func, or_
from sqlalchemy.exc import IntegrityError
@@ -19,6 +20,7 @@ from govoplan_core.core.events import (
PlatformEvent,
emit_platform_event,
)
from govoplan_core.core.concurrency import MissingPreconditionError, claim_revision
from govoplan_core.core.identity import (
CAPABILITY_IDENTITY_DIRECTORY,
IdentityDirectory,
@@ -57,6 +59,7 @@ from govoplan_core.core.postbox import (
PostboxDeliveryTemplateRef,
PostboxDirectoryEntryRef,
PostboxMessageAvailability,
PostboxMessageAuthoringRequest,
PostboxMessageRef,
PostboxMessageListState,
PostboxOrganizationFunctionTargetRef,
@@ -124,6 +127,49 @@ def _mapping(value: Mapping[str, object] | None) -> dict[str, object]:
return dict(value or {})
def _authoring_digest(
request: PostboxMessageAuthoringRequest,
*,
in_reply_to: PostboxMessage | None,
) -> str:
payload = {
"subject": request.subject.strip() or "(No subject)",
"body_text": request.body_text,
"classification": request.classification.strip().casefold(),
"in_reply_to_message_id": in_reply_to.id if in_reply_to else None,
"participants": [
{
"kind": item.kind,
"reference_type": item.reference_type,
"reference_id": item.reference_id,
"label": item.label,
"address": item.address,
}
for item in request.participants
],
"attachments": [
{
"reference_type": item.reference_type,
"reference_id": item.reference_id,
"name": item.name,
"media_type": item.media_type,
"size_bytes": item.size_bytes,
"digest": item.digest,
"metadata": dict(item.metadata),
}
for item in request.attachments
],
"metadata": dict(request.metadata),
}
encoded = json.dumps(
payload,
sort_keys=True,
separators=(",", ":"),
default=str,
).encode("utf-8")
return hashlib.sha256(encoded).hexdigest()
def _as_utc(value: datetime) -> datetime:
if value.tzinfo is None:
return value.replace(tzinfo=timezone.utc)
@@ -772,6 +818,268 @@ class PostboxService:
db.refresh(message)
return self._message_ref(message, account_id=actor.account_id)
def create_message(
self,
session: object,
*,
tenant_id: str,
postbox_id: str,
actor: PostboxActorRef,
request: PostboxMessageAuthoringRequest,
) -> PostboxMessageRef:
db = _session(session)
postbox = self._get_postbox(
db,
tenant_id=tenant_id,
postbox_id=postbox_id,
)
return self._author_message(
db,
postbox=postbox,
actor=actor,
request=request,
action="send",
in_reply_to=None,
)
def reply_to_message(
self,
session: object,
*,
tenant_id: str,
message_id: str,
actor: PostboxActorRef,
request: PostboxMessageAuthoringRequest,
) -> PostboxMessageRef:
db = _session(session)
parent = self._get_message(
db,
tenant_id=tenant_id,
message_id=message_id,
)
availability = _message_availability(parent)
if availability != "available":
raise PostboxError(
f"message_{availability}",
f"This Postbox message is {availability} and cannot be replied to.",
)
postbox = self._get_postbox(
db,
tenant_id=tenant_id,
postbox_id=parent.postbox_id,
)
classification = self._validate_classification(request.classification)
if not postbox_classification_allows(
classification,
parent.classification,
):
raise PostboxError(
"reply_classification_too_low",
"A reply cannot be classified below its parent message.",
)
return self._author_message(
db,
postbox=postbox,
actor=actor,
request=request,
action="reply",
in_reply_to=parent,
)
def _author_message(
self,
session: Session,
*,
postbox: Postbox,
actor: PostboxActorRef,
request: PostboxMessageAuthoringRequest,
action: Literal["send", "reply"],
in_reply_to: PostboxMessage | None,
) -> PostboxMessageRef:
classification = self._validate_classification(request.classification)
if not postbox_classification_allows(
postbox.classification,
classification,
):
raise PostboxError(
"classification_not_allowed",
"The message classification exceeds the Postbox classification.",
)
decision = self._access_decision(
postbox,
actor=actor,
action=action,
assignments=self._assignments_for_actor(
actor,
tenant_id=postbox.tenant_id,
),
holder_cache={},
classification=classification,
)
if not decision.allowed:
self._record_access_event(
session,
tenant_id=postbox.tenant_id,
postbox_id=postbox.id,
message_id=in_reply_to.id if in_reply_to else None,
actor=actor,
action=f"message.{action}",
outcome="denied",
reason_code=decision.reason_code,
assignment_id=decision.selected_assignment_id,
)
raise PostboxError("access_denied", decision.explanation)
authoring_key = request.idempotency_key.strip()
if not authoring_key:
raise PostboxError(
"idempotency_key_required",
"Message authoring requires an idempotency key.",
)
digest = _authoring_digest(request, in_reply_to=in_reply_to)
existing = (
session.query(PostboxMessage)
.options(
selectinload(PostboxMessage.participants),
selectinload(PostboxMessage.attachments),
selectinload(PostboxMessage.receipts),
)
.filter(
PostboxMessage.tenant_id == postbox.tenant_id,
PostboxMessage.postbox_id == postbox.id,
PostboxMessage.authoring_key == authoring_key,
)
.one_or_none()
)
if existing is not None:
authoring = _mapping(existing.metadata_).get("authoring")
existing_digest = (
str(authoring.get("digest"))
if isinstance(authoring, Mapping) and authoring.get("digest")
else None
)
if existing_digest != digest:
raise PostboxError(
"idempotency_conflict",
"The message idempotency key was already used for different content.",
)
return self._message_ref(existing, account_id=actor.account_id)
identity = (
self._identities.get_identity(actor.identity_id)
if actor.identity_id
else self._identities.identity_for_account(actor.account_id)
)
sender_label = (
identity.display_name
if identity is not None and identity.display_name
else actor.account_id
)
metadata = _mapping(request.metadata)
metadata["authoring"] = {
"digest": digest,
"account_id": actor.account_id,
"identity_id": actor.identity_id,
"assignment_id": decision.selected_assignment_id,
"action": action,
}
message = PostboxMessage(
tenant_id=postbox.tenant_id,
postbox_id=postbox.id,
subject=request.subject.strip() or "(No subject)",
body_text=request.body_text,
status="sent",
classification=classification,
sender_label=sender_label,
producer_module="postbox",
producer_resource_type="account_authored_message",
producer_resource_id=actor.account_id,
authoring_key=authoring_key,
in_reply_to_message_id=(in_reply_to.id if in_reply_to else None),
encryption_profile=postbox.encryption_profile,
key_epoch=postbox.key_epoch,
delivered_at=utc_now(),
metadata_=metadata,
)
message.participants.append(
PostboxParticipant(
tenant_id=postbox.tenant_id,
kind="author",
reference_type="account",
reference_id=actor.account_id,
label=sender_label,
position=0,
metadata_={"assignment_id": decision.selected_assignment_id},
)
)
for position, participant in enumerate(request.participants, start=1):
message.participants.append(
PostboxParticipant(
tenant_id=postbox.tenant_id,
kind=participant.kind,
reference_type=participant.reference_type,
reference_id=participant.reference_id,
label=participant.label,
address=participant.address,
position=position,
metadata_={},
)
)
for position, attachment in enumerate(request.attachments):
message.attachments.append(
PostboxAttachmentReference(
tenant_id=postbox.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,
position=position,
metadata_=_mapping(attachment.metadata),
)
)
session.add(message)
try:
session.flush()
except IntegrityError as exc:
raise PostboxError(
"idempotency_conflict",
"The message idempotency key was accepted concurrently.",
) from exc
self._record_access_event(
session,
tenant_id=postbox.tenant_id,
postbox_id=postbox.id,
message_id=message.id,
actor=actor,
action=f"message.{action}",
outcome="allowed",
reason_code=decision.reason_code,
assignment_id=decision.selected_assignment_id,
details={"in_reply_to_message_id": message.in_reply_to_message_id},
)
_publish_postbox_event(
session,
(
"postbox.message.replied.v1"
if in_reply_to is not None
else "postbox.message.authored.v1"
),
tenant_id=postbox.tenant_id,
resource_type="postbox_message",
resource_id=message.id,
postbox_id=postbox.id,
actor_type="account",
actor_id=actor.account_id,
payload={
"in_reply_to_message_id": message.in_reply_to_message_id,
"classification": classification,
"assignment_id": decision.selected_assignment_id,
},
)
return self._message_ref(message, account_id=actor.account_id)
# Capability: delivery
def deliver(
self,
@@ -2258,12 +2566,21 @@ class PostboxService:
tenant_id: str,
postbox_id: str,
actor_id: str | None,
expected_revision: int,
) -> Postbox:
postbox = self._get_postbox(
session,
tenant_id=tenant_id,
postbox_id=postbox_id,
)
self._claim_resource_revision(
session,
model=Postbox,
resource=postbox,
resource_type="postbox",
tenant_id=tenant_id,
expected_revision=expected_revision,
)
postbox.status = "archived"
postbox.archived_at = utc_now()
postbox.address_record.status = "archived"
@@ -2405,6 +2722,7 @@ class PostboxService:
classification: str,
allow_vacant_delivery: bool,
actor_id: str | None,
expected_revision: int,
routing_policy: Mapping[str, object] | None = None,
) -> PostboxTemplate:
classification = self._validate_classification(classification)
@@ -2418,6 +2736,14 @@ class PostboxService:
"template_retired",
"A retired Postbox template cannot be revised.",
)
self._claim_resource_revision(
session,
model=PostboxTemplate,
resource=template,
resource_type="postbox_template",
tenant_id=tenant_id,
expected_revision=expected_revision,
)
self._validate_scope(
tenant_id=tenant_id,
scope_kind=scope_kind,
@@ -2470,12 +2796,21 @@ class PostboxService:
template_id: str,
revision_number: int | None,
actor_id: str | None,
expected_revision: int,
) -> PostboxTemplate:
template = self._get_template(
session,
tenant_id=tenant_id,
template_id=template_id,
)
self._claim_resource_revision(
session,
model=PostboxTemplate,
resource=template,
resource_type="postbox_template",
tenant_id=tenant_id,
expected_revision=expected_revision,
)
revision = next(
(
item
@@ -2519,12 +2854,21 @@ class PostboxService:
tenant_id: str,
template_id: str,
actor_id: str | None,
expected_revision: int,
) -> PostboxTemplate:
template = self._get_template(
session,
tenant_id=tenant_id,
template_id=template_id,
)
self._claim_resource_revision(
session,
model=PostboxTemplate,
resource=template,
resource_type="postbox_template",
tenant_id=tenant_id,
expected_revision=expected_revision,
)
template.status = "retired"
template.retired_at = utc_now()
template.updated_by = actor_id
@@ -2773,6 +3117,7 @@ class PostboxService:
name: str,
is_default: bool,
postbox_ids: Sequence[str],
expected_revision: int | None = None,
) -> PostboxGrouping:
grouping = (
session.query(PostboxGrouping)
@@ -2788,6 +3133,16 @@ class PostboxService:
)
if grouping_id and grouping is None:
raise PostboxError("grouping_not_found", "Postbox grouping not found.")
if grouping is not None:
self._claim_resource_revision(
session,
model=PostboxGrouping,
resource=grouping,
resource_type="postbox_grouping",
tenant_id=tenant_id,
expected_revision=expected_revision,
account_id=actor.account_id,
)
existing_ids = tuple(
source.postbox_id for source in grouping.sources
) if grouping is not None else ()
@@ -2822,6 +3177,7 @@ class PostboxService:
settings={},
)
session.add(grouping)
session.flush()
else:
grouping.name = name.strip()
grouping.is_default = is_default
@@ -2833,8 +3189,16 @@ class PostboxService:
PostboxGrouping.tenant_id == tenant_id,
PostboxGrouping.account_id == actor.account_id,
PostboxGrouping.id != grouping.id,
PostboxGrouping.is_default.is_(True),
)
.update(
{
PostboxGrouping.is_default: False,
PostboxGrouping.resource_revision: (
PostboxGrouping.resource_revision + 1
),
}
)
.update({PostboxGrouping.is_default: False})
)
for position, postbox_id in enumerate(saved_ids):
grouping.sources.append(
@@ -2854,6 +3218,7 @@ class PostboxService:
tenant_id: str,
actor: PostboxActorRef,
grouping_id: str,
expected_revision: int,
) -> None:
grouping = (
session.query(PostboxGrouping)
@@ -2866,10 +3231,53 @@ class PostboxService:
)
if grouping is None:
raise PostboxError("grouping_not_found", "Postbox grouping not found.")
self._claim_resource_revision(
session,
model=PostboxGrouping,
resource=grouping,
resource_type="postbox_grouping",
tenant_id=tenant_id,
expected_revision=expected_revision,
account_id=actor.account_id,
)
session.delete(grouping)
session.flush()
# Internal helpers
def _claim_resource_revision(
self,
session: Session,
*,
model: type[Any],
resource: Any,
resource_type: str,
tenant_id: str,
expected_revision: int | None,
account_id: str | None = None,
) -> int:
if expected_revision is None:
raise MissingPreconditionError(
resource_type=resource_type,
resource_id=resource.id,
)
filters = [
model.id == resource.id,
model.tenant_id == tenant_id,
]
if account_id is not None:
filters.append(model.account_id == account_id)
next_revision = claim_revision(
session,
model=model,
filters=filters,
revision_attribute="resource_revision",
expected_revision=expected_revision,
resource_type=resource_type,
resource_id=resource.id,
)
resource.resource_revision = next_revision
return next_revision
def _get_postbox(
self,
session: Session,
@@ -3282,6 +3690,8 @@ class PostboxService:
holder_count=holder_count,
vacant=holder_count == 0,
access=decision,
resource_revision=postbox.resource_revision,
etag=postbox.strong_etag,
)
def _message_ref(
@@ -3318,6 +3728,8 @@ class PostboxService:
producer_module=message.producer_module,
producer_resource_type=message.producer_resource_type,
producer_resource_id=message.producer_resource_id,
in_reply_to_message_id=message.in_reply_to_message_id,
replaces_message_id=message.replaces_message_id,
encryption_profile=message.encryption_profile,
key_epoch=message.key_epoch,
ciphertext_ref=message.ciphertext_ref,