Define governed platform event envelope
This commit is contained in:
@@ -58,6 +58,16 @@ Every `PlatformEvent` has:
|
|||||||
- `correlation_id`: stable ID for the whole request, workflow, or job.
|
- `correlation_id`: stable ID for the whole request, workflow, or job.
|
||||||
- `causation_id`: the event ID or external operation ID that caused this
|
- `causation_id`: the event ID or external operation ID that caused this
|
||||||
event.
|
event.
|
||||||
|
- `actor`: optional typed actor reference, for example user, API key, system
|
||||||
|
actor, delegated actor, or installer daemon.
|
||||||
|
- `tenant`: optional tenant reference when the event is tenant-scoped.
|
||||||
|
- `subject`: optional typed subject reference for the person, organization,
|
||||||
|
account, case, campaign, or other entity the event is about.
|
||||||
|
- `resource`: optional typed resource reference for the object changed or
|
||||||
|
observed by the event.
|
||||||
|
- `classification`: payload sensitivity, currently `public`, `internal`,
|
||||||
|
`confidential`, or `restricted`.
|
||||||
|
- `payload`: JSON-serializable module-owned event details.
|
||||||
|
|
||||||
The FastAPI app factory creates an event context for every request. It accepts
|
The FastAPI app factory creates an event context for every request. It accepts
|
||||||
`X-Correlation-ID` or `X-Request-ID` when the value is a compact safe trace ID,
|
`X-Correlation-ID` or `X-Request-ID` when the value is a compact safe trace ID,
|
||||||
@@ -174,7 +184,10 @@ Campaign:
|
|||||||
- Payloads must be JSON-serializable.
|
- Payloads must be JSON-serializable.
|
||||||
- Use stable IDs, not ORM objects.
|
- Use stable IDs, not ORM objects.
|
||||||
- Include tenant ID when tenant-scoped.
|
- Include tenant ID when tenant-scoped.
|
||||||
- Include actor/principal references only as DTOs or primitive IDs.
|
- Include actor/principal, subject, and resource references only as typed DTOs
|
||||||
|
or primitive IDs.
|
||||||
|
- Set `classification` to the highest sensitivity needed by the envelope or
|
||||||
|
payload, not the lowest sensitivity of any individual field.
|
||||||
- Do not include secrets, raw message bodies, full recipient lists, or file
|
- Do not include secrets, raw message bodies, full recipient lists, or file
|
||||||
content.
|
content.
|
||||||
- Put large evidence in owning module storage and reference it by ID.
|
- Put large evidence in owning module storage and reference it by ID.
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ from contextvars import ContextVar
|
|||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
import re
|
import re
|
||||||
from typing import Any
|
from typing import Any, Literal
|
||||||
import uuid
|
import uuid
|
||||||
|
|
||||||
|
|
||||||
@@ -31,6 +31,39 @@ class EventTrace:
|
|||||||
causation_id: str | None = None
|
causation_id: str | None = None
|
||||||
|
|
||||||
|
|
||||||
|
EventClassification = Literal["public", "internal", "confidential", "restricted"]
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class EventActorRef:
|
||||||
|
type: str
|
||||||
|
id: str | None = None
|
||||||
|
label: str | None = None
|
||||||
|
|
||||||
|
def to_dict(self) -> dict[str, Any]:
|
||||||
|
return _compact_dict({"type": self.type, "id": self.id, "label": self.label})
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class EventTenantRef:
|
||||||
|
id: str
|
||||||
|
slug: str | None = None
|
||||||
|
label: str | None = None
|
||||||
|
|
||||||
|
def to_dict(self) -> dict[str, Any]:
|
||||||
|
return _compact_dict({"id": self.id, "slug": self.slug, "label": self.label})
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class EventObjectRef:
|
||||||
|
type: str
|
||||||
|
id: str | None = None
|
||||||
|
label: str | None = None
|
||||||
|
|
||||||
|
def to_dict(self) -> dict[str, Any]:
|
||||||
|
return _compact_dict({"type": self.type, "id": self.id, "label": self.label})
|
||||||
|
|
||||||
|
|
||||||
_current_trace: ContextVar[EventTrace | None] = ContextVar("govoplan_event_trace", default=None)
|
_current_trace: ContextVar[EventTrace | None] = ContextVar("govoplan_event_trace", default=None)
|
||||||
|
|
||||||
|
|
||||||
@@ -43,6 +76,27 @@ class PlatformEvent:
|
|||||||
event_id: str = field(default_factory=new_event_id)
|
event_id: str = field(default_factory=new_event_id)
|
||||||
correlation_id: str | None = None
|
correlation_id: str | None = None
|
||||||
causation_id: str | None = None
|
causation_id: str | None = None
|
||||||
|
actor: EventActorRef | None = None
|
||||||
|
tenant: EventTenantRef | None = None
|
||||||
|
subject: EventObjectRef | None = None
|
||||||
|
resource: EventObjectRef | None = None
|
||||||
|
classification: EventClassification = "internal"
|
||||||
|
|
||||||
|
def to_dict(self) -> dict[str, Any]:
|
||||||
|
return {
|
||||||
|
"type": self.type,
|
||||||
|
"module_id": self.module_id,
|
||||||
|
"payload": dict(self.payload),
|
||||||
|
"occurred_at": self.occurred_at.isoformat(),
|
||||||
|
"event_id": self.event_id,
|
||||||
|
"correlation_id": self.correlation_id,
|
||||||
|
"causation_id": self.causation_id,
|
||||||
|
"actor": self.actor.to_dict() if self.actor else None,
|
||||||
|
"tenant": self.tenant.to_dict() if self.tenant else None,
|
||||||
|
"subject": self.subject.to_dict() if self.subject else None,
|
||||||
|
"resource": self.resource.to_dict() if self.resource else None,
|
||||||
|
"classification": self.classification,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
EventHandler = Callable[[PlatformEvent], None]
|
EventHandler = Callable[[PlatformEvent], None]
|
||||||
@@ -90,6 +144,11 @@ def ensure_event_trace(event: PlatformEvent) -> PlatformEvent:
|
|||||||
event_id=event.event_id,
|
event_id=event.event_id,
|
||||||
correlation_id=trace.correlation_id,
|
correlation_id=trace.correlation_id,
|
||||||
causation_id=trace.causation_id,
|
causation_id=trace.causation_id,
|
||||||
|
actor=event.actor,
|
||||||
|
tenant=event.tenant,
|
||||||
|
subject=event.subject,
|
||||||
|
resource=event.resource,
|
||||||
|
classification=event.classification,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -107,3 +166,7 @@ class EventBus:
|
|||||||
handler(traced)
|
handler(traced)
|
||||||
for handler in self._subscribers.get("*", ()):
|
for handler in self._subscribers.get("*", ()):
|
||||||
handler(traced)
|
handler(traced)
|
||||||
|
|
||||||
|
|
||||||
|
def _compact_dict(value: Mapping[str, Any]) -> dict[str, Any]:
|
||||||
|
return {key: item for key, item in value.items() if item is not None}
|
||||||
|
|||||||
@@ -9,7 +9,16 @@ from fastapi import APIRouter
|
|||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
from govoplan_core.audit.logging import audit_event, audit_operation_context
|
from govoplan_core.audit.logging import audit_event, audit_operation_context
|
||||||
from govoplan_core.core.events import EventBus, PlatformEvent, current_event_trace, event_context, normalize_trace_id
|
from govoplan_core.core.events import (
|
||||||
|
EventActorRef,
|
||||||
|
EventBus,
|
||||||
|
EventObjectRef,
|
||||||
|
EventTenantRef,
|
||||||
|
PlatformEvent,
|
||||||
|
current_event_trace,
|
||||||
|
event_context,
|
||||||
|
normalize_trace_id,
|
||||||
|
)
|
||||||
from govoplan_core.core.registry import PlatformRegistry
|
from govoplan_core.core.registry import PlatformRegistry
|
||||||
from govoplan_core.db.base import Base
|
from govoplan_core.db.base import Base
|
||||||
from govoplan_core.server.fastapi import create_govoplan_app
|
from govoplan_core.server.fastapi import create_govoplan_app
|
||||||
@@ -37,6 +46,60 @@ class CoreEventTests(unittest.TestCase):
|
|||||||
self.assertEqual("corr-1", child.correlation_id)
|
self.assertEqual("corr-1", child.correlation_id)
|
||||||
self.assertEqual(parent.event_id, child.causation_id)
|
self.assertEqual(parent.event_id, child.causation_id)
|
||||||
|
|
||||||
|
def test_platform_event_serializes_governed_envelope_fields(self) -> None:
|
||||||
|
event = PlatformEvent(
|
||||||
|
type="case.permit.updated",
|
||||||
|
module_id="cases",
|
||||||
|
payload={"status": "approved"},
|
||||||
|
event_id="event-1",
|
||||||
|
correlation_id="corr-1",
|
||||||
|
causation_id="cause-1",
|
||||||
|
actor=EventActorRef(type="user", id="user-1", label="Case worker"),
|
||||||
|
tenant=EventTenantRef(id="tenant-1", slug="city", label="City Office"),
|
||||||
|
subject=EventObjectRef(type="person", id="person-1", label="Applicant"),
|
||||||
|
resource=EventObjectRef(type="permit", id="permit-1", label="Permit"),
|
||||||
|
classification="confidential",
|
||||||
|
)
|
||||||
|
|
||||||
|
payload = event.to_dict()
|
||||||
|
|
||||||
|
self.assertEqual("case.permit.updated", payload["type"])
|
||||||
|
self.assertEqual("cases", payload["module_id"])
|
||||||
|
self.assertEqual({"status": "approved"}, payload["payload"])
|
||||||
|
self.assertEqual("event-1", payload["event_id"])
|
||||||
|
self.assertEqual("corr-1", payload["correlation_id"])
|
||||||
|
self.assertEqual("cause-1", payload["causation_id"])
|
||||||
|
self.assertEqual({"type": "user", "id": "user-1", "label": "Case worker"}, payload["actor"])
|
||||||
|
self.assertEqual({"id": "tenant-1", "slug": "city", "label": "City Office"}, payload["tenant"])
|
||||||
|
self.assertEqual({"type": "person", "id": "person-1", "label": "Applicant"}, payload["subject"])
|
||||||
|
self.assertEqual({"type": "permit", "id": "permit-1", "label": "Permit"}, payload["resource"])
|
||||||
|
self.assertEqual("confidential", payload["classification"])
|
||||||
|
|
||||||
|
def test_event_bus_preserves_envelope_fields_when_adding_trace(self) -> None:
|
||||||
|
bus = EventBus()
|
||||||
|
seen: list[PlatformEvent] = []
|
||||||
|
event = PlatformEvent(
|
||||||
|
type="files.file.uploaded",
|
||||||
|
module_id="files",
|
||||||
|
actor=EventActorRef(type="api_key", id="key-1"),
|
||||||
|
tenant=EventTenantRef(id="tenant-1"),
|
||||||
|
subject=EventObjectRef(type="file", id="file-1"),
|
||||||
|
resource=EventObjectRef(type="folder", id="folder-1"),
|
||||||
|
classification="restricted",
|
||||||
|
)
|
||||||
|
|
||||||
|
bus.subscribe("files.file.uploaded", seen.append)
|
||||||
|
bus.publish(event)
|
||||||
|
|
||||||
|
self.assertEqual(1, len(seen))
|
||||||
|
traced = seen[0]
|
||||||
|
self.assertIsNotNone(traced.correlation_id)
|
||||||
|
self.assertEqual(event.actor, traced.actor)
|
||||||
|
self.assertEqual(event.tenant, traced.tenant)
|
||||||
|
self.assertEqual(event.subject, traced.subject)
|
||||||
|
self.assertEqual(event.resource, traced.resource)
|
||||||
|
self.assertEqual("restricted", traced.classification)
|
||||||
|
|
||||||
def test_request_correlation_middleware_sets_event_context_and_response_header(self) -> None:
|
def test_request_correlation_middleware_sets_event_context_and_response_header(self) -> None:
|
||||||
router = APIRouter()
|
router = APIRouter()
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user