660 lines
20 KiB
Python
660 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Callable, Mapping, Sequence
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any, cast
|
|
|
|
from sqlalchemy import delete, func, or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_audit.backend.db.models import (
|
|
AuditOutboxDelivery,
|
|
AuditOutboxEvent,
|
|
)
|
|
from govoplan_core.core.events import (
|
|
DurableEventConsumer,
|
|
EventActorRef,
|
|
EventClassification,
|
|
EventObjectRef,
|
|
EventTenantRef,
|
|
PlatformEvent,
|
|
ensure_event_trace,
|
|
publish_platform_event,
|
|
)
|
|
from govoplan_core.core.institutional import GovernedContextEnvelope
|
|
|
|
EventDispatcher = Callable[[PlatformEvent], None]
|
|
|
|
|
|
class SqlAuditOutbox:
|
|
def __init__(self, *, max_attempts: int = 8) -> None:
|
|
self._max_attempts = max(1, min(int(max_attempts), 100))
|
|
|
|
def enqueue(self, session: object, event: PlatformEvent) -> AuditOutboxEvent:
|
|
db = _session(session)
|
|
traced = ensure_event_trace(event)
|
|
existing = db.scalar(
|
|
select(AuditOutboxEvent).where(
|
|
AuditOutboxEvent.event_id == traced.event_id
|
|
)
|
|
)
|
|
if existing is not None:
|
|
if existing.payload != traced.to_dict():
|
|
raise ValueError(
|
|
"A different platform event already uses this event id"
|
|
)
|
|
return existing
|
|
item = AuditOutboxEvent(
|
|
event_id=traced.event_id,
|
|
event_type=traced.type,
|
|
module_id=traced.module_id,
|
|
correlation_id=traced.correlation_id,
|
|
causation_id=traced.causation_id,
|
|
classification=traced.classification,
|
|
payload=traced.to_dict(),
|
|
status="pending",
|
|
)
|
|
db.add(item)
|
|
return item
|
|
|
|
def dispatch_pending(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str | None = None,
|
|
tenantless_only: bool = False,
|
|
consumers: Sequence[DurableEventConsumer] = (),
|
|
observer: EventDispatcher | None = publish_platform_event,
|
|
dispatcher: EventDispatcher | None = None,
|
|
limit: int = 100,
|
|
) -> dict[str, int]:
|
|
if tenant_id is not None and tenantless_only:
|
|
raise ValueError(
|
|
"Tenant and tenantless event filters are mutually exclusive"
|
|
)
|
|
db = _session(session)
|
|
now = datetime.now(timezone.utc)
|
|
consumers_by_id = _consumer_map(consumers)
|
|
query = (
|
|
db.query(AuditOutboxEvent)
|
|
.filter(
|
|
AuditOutboxEvent.status.in_(
|
|
("pending", "failed", "retrying")
|
|
),
|
|
or_(AuditOutboxEvent.next_attempt_at.is_(None), AuditOutboxEvent.next_attempt_at <= now),
|
|
)
|
|
)
|
|
if tenant_id:
|
|
query = query.filter(
|
|
AuditOutboxEvent.payload["tenant"]["id"].as_string()
|
|
== tenant_id
|
|
)
|
|
elif tenantless_only:
|
|
query = query.filter(
|
|
AuditOutboxEvent.payload["tenant"]["id"]
|
|
.as_string()
|
|
.is_(None)
|
|
)
|
|
rows = (
|
|
query.order_by(AuditOutboxEvent.created_at.asc(), AuditOutboxEvent.id.asc())
|
|
.with_for_update(skip_locked=True)
|
|
.limit(max(1, min(int(limit), 500)))
|
|
.all()
|
|
)
|
|
counts = {
|
|
"selected": len(rows),
|
|
"delivered": 0,
|
|
"retrying": 0,
|
|
"quarantined": 0,
|
|
"dispatched": 0,
|
|
"observer_failed": 0,
|
|
}
|
|
effective_observer = dispatcher or observer
|
|
for row in rows:
|
|
event = _event_from_payload(row.payload)
|
|
deliveries = _event_deliveries(
|
|
db,
|
|
row=row,
|
|
event=event,
|
|
consumers=consumers_by_id.values(),
|
|
)
|
|
_dispatch_event_deliveries(
|
|
event,
|
|
deliveries=deliveries,
|
|
consumers_by_id=consumers_by_id,
|
|
now=now,
|
|
max_attempts=self._max_attempts,
|
|
counts=counts,
|
|
)
|
|
_finish_event_dispatch(
|
|
row,
|
|
deliveries=deliveries,
|
|
observer=effective_observer,
|
|
event=event,
|
|
now=now,
|
|
counts=counts,
|
|
)
|
|
db.flush()
|
|
return counts
|
|
|
|
def replay_delivery(
|
|
self,
|
|
session: object,
|
|
*,
|
|
event_id: str,
|
|
consumer_id: str,
|
|
operator_id: str,
|
|
reason: str,
|
|
) -> dict[str, object]:
|
|
db = _session(session)
|
|
clean_reason = reason.strip()
|
|
clean_operator_id = operator_id.strip()
|
|
if not clean_reason or len(clean_reason) > 2000:
|
|
raise ValueError(
|
|
"Replay reason must contain between 1 and 2000 characters"
|
|
)
|
|
if not clean_operator_id or len(clean_operator_id) > 128:
|
|
raise ValueError("Replay operator id is invalid")
|
|
row = db.scalar(
|
|
select(AuditOutboxDelivery)
|
|
.join(
|
|
AuditOutboxEvent,
|
|
AuditOutboxEvent.id
|
|
== AuditOutboxDelivery.outbox_event_id,
|
|
)
|
|
.where(
|
|
AuditOutboxEvent.event_id == event_id,
|
|
AuditOutboxDelivery.consumer_id == consumer_id,
|
|
)
|
|
.with_for_update()
|
|
)
|
|
if row is None:
|
|
raise LookupError("Platform event delivery was not found")
|
|
if row.status not in {"retrying", "quarantined"}:
|
|
raise ValueError(
|
|
"Only retrying or quarantined deliveries can be replayed"
|
|
)
|
|
now = datetime.now(timezone.utc)
|
|
row.status = "pending"
|
|
row.attempts = 0
|
|
row.next_attempt_at = now
|
|
row.quarantined_at = None
|
|
row.last_error = None
|
|
row.replay_count += 1
|
|
row.last_replayed_at = now
|
|
row.last_replayed_by = clean_operator_id
|
|
row.last_replay_reason = clean_reason
|
|
event_row = db.get(AuditOutboxEvent, row.outbox_event_id)
|
|
if event_row is None:
|
|
raise LookupError("Platform event envelope was not found")
|
|
event_row.status = "pending"
|
|
event_row.next_attempt_at = now
|
|
event_row.last_error = None
|
|
event_row.dispatched_at = None
|
|
db.flush()
|
|
return _delivery_state(row, event_id=event_row.event_id)
|
|
|
|
def purge_terminal(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str | None = None,
|
|
tenantless_only: bool = False,
|
|
before: datetime,
|
|
limit: int = 500,
|
|
) -> dict[str, int]:
|
|
if tenant_id is not None and tenantless_only:
|
|
raise ValueError(
|
|
"Tenant and tenantless event filters are mutually exclusive"
|
|
)
|
|
db = _session(session)
|
|
clauses = [
|
|
AuditOutboxEvent.status == "dispatched",
|
|
AuditOutboxEvent.dispatched_at.is_not(None),
|
|
AuditOutboxEvent.dispatched_at < before,
|
|
]
|
|
if tenant_id:
|
|
clauses.append(
|
|
AuditOutboxEvent.payload["tenant"]["id"].as_string()
|
|
== tenant_id
|
|
)
|
|
elif tenantless_only:
|
|
clauses.append(
|
|
AuditOutboxEvent.payload["tenant"]["id"]
|
|
.as_string()
|
|
.is_(None)
|
|
)
|
|
ids = tuple(
|
|
db.scalars(
|
|
select(AuditOutboxEvent.id)
|
|
.where(*clauses)
|
|
.order_by(
|
|
AuditOutboxEvent.dispatched_at,
|
|
AuditOutboxEvent.id,
|
|
)
|
|
.limit(max(1, min(int(limit), 5000)))
|
|
)
|
|
)
|
|
if ids:
|
|
db.execute(
|
|
delete(AuditOutboxEvent).where(
|
|
AuditOutboxEvent.id.in_(ids)
|
|
)
|
|
)
|
|
db.flush()
|
|
return {"deleted": len(ids)}
|
|
|
|
def delivery_metrics(
|
|
self,
|
|
session: object,
|
|
) -> dict[str, object]:
|
|
db = _session(session)
|
|
event_counts = {
|
|
str(status): int(count)
|
|
for status, count in db.execute(
|
|
select(
|
|
AuditOutboxEvent.status,
|
|
func.count(AuditOutboxEvent.id),
|
|
).group_by(AuditOutboxEvent.status)
|
|
)
|
|
}
|
|
delivery_counts = {
|
|
str(status): int(count)
|
|
for status, count in db.execute(
|
|
select(
|
|
AuditOutboxDelivery.status,
|
|
func.count(AuditOutboxDelivery.id),
|
|
).group_by(AuditOutboxDelivery.status)
|
|
)
|
|
}
|
|
consumer_counts = {
|
|
str(consumer_id): {
|
|
str(status): int(count)
|
|
for status, count in values
|
|
}
|
|
for consumer_id, values in _consumer_delivery_counts(db).items()
|
|
}
|
|
oldest_due = db.scalar(
|
|
select(func.min(AuditOutboxDelivery.created_at)).where(
|
|
AuditOutboxDelivery.status.in_(
|
|
("pending", "retrying")
|
|
)
|
|
)
|
|
)
|
|
return {
|
|
"events": event_counts,
|
|
"deliveries": delivery_counts,
|
|
"consumers": consumer_counts,
|
|
"oldest_due_at": (
|
|
oldest_due.isoformat()
|
|
if isinstance(oldest_due, datetime)
|
|
else None
|
|
),
|
|
}
|
|
|
|
|
|
def _consumer_map(
|
|
consumers: Sequence[DurableEventConsumer],
|
|
) -> dict[str, DurableEventConsumer]:
|
|
result: dict[str, DurableEventConsumer] = {}
|
|
for consumer in consumers:
|
|
if consumer.consumer_id in result:
|
|
raise ValueError(
|
|
f"Duplicate durable event consumer: {consumer.consumer_id}"
|
|
)
|
|
result[consumer.consumer_id] = consumer
|
|
return result
|
|
|
|
|
|
def _event_deliveries(
|
|
session: Session,
|
|
*,
|
|
row: AuditOutboxEvent,
|
|
event: PlatformEvent,
|
|
consumers: Sequence[DurableEventConsumer],
|
|
) -> list[AuditOutboxDelivery]:
|
|
existing = {
|
|
delivery.consumer_id: delivery
|
|
for delivery in session.scalars(
|
|
select(AuditOutboxDelivery).where(
|
|
AuditOutboxDelivery.outbox_event_id == row.id
|
|
)
|
|
)
|
|
}
|
|
for consumer in consumers:
|
|
if (
|
|
consumer.consumer_id in existing
|
|
or not consumer.accepts(event)
|
|
):
|
|
continue
|
|
delivery = AuditOutboxDelivery(
|
|
outbox_event_id=row.id,
|
|
consumer_id=consumer.consumer_id,
|
|
delivery_key=consumer.delivery_key(event),
|
|
policy_decision_ref=consumer.policy_decision_ref,
|
|
status="pending",
|
|
)
|
|
session.add(delivery)
|
|
existing[consumer.consumer_id] = delivery
|
|
session.flush()
|
|
return sorted(
|
|
existing.values(),
|
|
key=lambda item: (item.created_at, item.id),
|
|
)
|
|
|
|
|
|
def _dispatch_event_deliveries(
|
|
event: PlatformEvent,
|
|
*,
|
|
deliveries: Sequence[AuditOutboxDelivery],
|
|
consumers_by_id: Mapping[str, DurableEventConsumer],
|
|
now: datetime,
|
|
max_attempts: int,
|
|
counts: dict[str, int],
|
|
) -> None:
|
|
for delivery in deliveries:
|
|
if not _delivery_is_due(delivery, now=now):
|
|
continue
|
|
consumer = consumers_by_id.get(delivery.consumer_id)
|
|
if consumer is None:
|
|
_record_delivery_failure(
|
|
delivery,
|
|
error="Durable event consumer is not registered",
|
|
now=now,
|
|
max_attempts=max_attempts,
|
|
counts=counts,
|
|
)
|
|
continue
|
|
if not consumer.accepts(event):
|
|
_quarantine_delivery(
|
|
delivery,
|
|
error=(
|
|
"The current durable subscription no longer permits "
|
|
"this event"
|
|
),
|
|
now=now,
|
|
counts=counts,
|
|
)
|
|
continue
|
|
if (
|
|
event.classification in {"confidential", "restricted"}
|
|
and delivery.policy_decision_ref
|
|
!= consumer.policy_decision_ref
|
|
):
|
|
_quarantine_delivery(
|
|
delivery,
|
|
error=(
|
|
"The policy decision for this classified event "
|
|
"subscription changed"
|
|
),
|
|
now=now,
|
|
counts=counts,
|
|
)
|
|
continue
|
|
try:
|
|
consumer.handler(event, delivery.delivery_key)
|
|
except Exception as exc: # noqa: BLE001 - failures must be persisted.
|
|
_record_delivery_failure(
|
|
delivery,
|
|
error=str(exc),
|
|
now=now,
|
|
max_attempts=max_attempts,
|
|
counts=counts,
|
|
)
|
|
continue
|
|
delivery.status = "delivered"
|
|
delivery.attempts += 1
|
|
delivery.delivered_at = now
|
|
delivery.next_attempt_at = None
|
|
delivery.quarantined_at = None
|
|
delivery.last_error = None
|
|
counts["delivered"] += 1
|
|
|
|
|
|
def _delivery_is_due(
|
|
delivery: AuditOutboxDelivery,
|
|
*,
|
|
now: datetime,
|
|
) -> bool:
|
|
if delivery.status not in {"pending", "retrying"}:
|
|
return False
|
|
if delivery.next_attempt_at is None:
|
|
return True
|
|
return _as_utc(delivery.next_attempt_at) <= now
|
|
|
|
|
|
def _record_delivery_failure(
|
|
delivery: AuditOutboxDelivery,
|
|
*,
|
|
error: str,
|
|
now: datetime,
|
|
max_attempts: int,
|
|
counts: dict[str, int],
|
|
) -> None:
|
|
delivery.attempts += 1
|
|
delivery.last_error = _bounded_error(error)
|
|
if delivery.attempts >= max_attempts:
|
|
_quarantine_delivery(
|
|
delivery,
|
|
error=delivery.last_error,
|
|
now=now,
|
|
counts=counts,
|
|
)
|
|
return
|
|
delivery.status = "retrying"
|
|
delivery.next_attempt_at = now + _retry_delay(delivery.attempts)
|
|
counts["retrying"] += 1
|
|
|
|
|
|
def _quarantine_delivery(
|
|
delivery: AuditOutboxDelivery,
|
|
*,
|
|
error: str,
|
|
now: datetime,
|
|
counts: dict[str, int],
|
|
) -> None:
|
|
delivery.status = "quarantined"
|
|
delivery.quarantined_at = now
|
|
delivery.next_attempt_at = None
|
|
delivery.last_error = _bounded_error(error)
|
|
counts["quarantined"] += 1
|
|
|
|
|
|
def _finish_event_dispatch(
|
|
row: AuditOutboxEvent,
|
|
*,
|
|
deliveries: Sequence[AuditOutboxDelivery],
|
|
observer: EventDispatcher | None,
|
|
event: PlatformEvent,
|
|
now: datetime,
|
|
counts: dict[str, int],
|
|
) -> None:
|
|
row.attempts += 1
|
|
quarantined = [
|
|
item for item in deliveries
|
|
if item.status == "quarantined"
|
|
]
|
|
outstanding = [
|
|
item for item in deliveries
|
|
if item.status in {"pending", "retrying"}
|
|
]
|
|
if quarantined:
|
|
row.status = "quarantined"
|
|
row.next_attempt_at = None
|
|
row.last_error = quarantined[0].last_error
|
|
return
|
|
if outstanding:
|
|
row.status = "retrying"
|
|
due_times = [
|
|
item.next_attempt_at
|
|
for item in outstanding
|
|
if item.next_attempt_at is not None
|
|
]
|
|
row.next_attempt_at = min(due_times) if due_times else now
|
|
row.last_error = next(
|
|
(
|
|
item.last_error
|
|
for item in outstanding
|
|
if item.last_error
|
|
),
|
|
None,
|
|
)
|
|
return
|
|
if observer is not None:
|
|
try:
|
|
observer(event)
|
|
except Exception as exc: # noqa: BLE001 - observers are non-durable.
|
|
counts["observer_failed"] += 1
|
|
row.last_error = _bounded_error(
|
|
f"Non-durable observer failed: {exc}"
|
|
)
|
|
else:
|
|
row.last_error = None
|
|
else:
|
|
row.last_error = None
|
|
row.status = "dispatched"
|
|
row.dispatched_at = now
|
|
row.next_attempt_at = None
|
|
counts["dispatched"] += 1
|
|
|
|
|
|
def _delivery_state(
|
|
delivery: AuditOutboxDelivery,
|
|
*,
|
|
event_id: str,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"event_id": event_id,
|
|
"consumer_id": delivery.consumer_id,
|
|
"delivery_key": delivery.delivery_key,
|
|
"status": delivery.status,
|
|
"attempts": delivery.attempts,
|
|
"replay_count": delivery.replay_count,
|
|
"last_replayed_at": delivery.last_replayed_at,
|
|
"last_replayed_by": delivery.last_replayed_by,
|
|
"last_replay_reason": delivery.last_replay_reason,
|
|
"last_error": delivery.last_error,
|
|
}
|
|
|
|
|
|
def _consumer_delivery_counts(
|
|
session: Session,
|
|
) -> dict[str, list[tuple[str, int]]]:
|
|
result: dict[str, list[tuple[str, int]]] = {}
|
|
for consumer_id, status, count in session.execute(
|
|
select(
|
|
AuditOutboxDelivery.consumer_id,
|
|
AuditOutboxDelivery.status,
|
|
func.count(AuditOutboxDelivery.id),
|
|
).group_by(
|
|
AuditOutboxDelivery.consumer_id,
|
|
AuditOutboxDelivery.status,
|
|
)
|
|
):
|
|
result.setdefault(str(consumer_id), []).append(
|
|
(str(status), int(count))
|
|
)
|
|
return result
|
|
|
|
|
|
def enqueue_platform_event(session: object, event: PlatformEvent) -> AuditOutboxEvent:
|
|
return SqlAuditOutbox().enqueue(session, event)
|
|
|
|
|
|
def dispatch_pending_platform_events(
|
|
session: object,
|
|
*,
|
|
dispatcher: EventDispatcher = publish_platform_event,
|
|
limit: int = 100,
|
|
) -> dict[str, int]:
|
|
return SqlAuditOutbox().dispatch_pending(session, dispatcher=dispatcher, limit=limit)
|
|
|
|
|
|
def _retry_delay(attempts: int) -> timedelta:
|
|
seconds = min(300, max(1, 2 ** max(0, attempts - 1)))
|
|
return timedelta(seconds=seconds)
|
|
|
|
|
|
def _as_utc(value: datetime) -> datetime:
|
|
if value.tzinfo is None:
|
|
return value.replace(tzinfo=timezone.utc)
|
|
return value.astimezone(timezone.utc)
|
|
|
|
|
|
def _bounded_error(value: str) -> str:
|
|
clean = value.strip() or "Unknown durable event delivery failure"
|
|
return clean[:4000]
|
|
|
|
|
|
def _event_from_payload(payload: Mapping[str, Any]) -> PlatformEvent:
|
|
return PlatformEvent(
|
|
type=str(payload["type"]),
|
|
module_id=str(payload["module_id"]),
|
|
payload=_mapping(payload.get("payload")),
|
|
occurred_at=_datetime(payload.get("occurred_at")),
|
|
event_id=str(payload["event_id"]),
|
|
correlation_id=_optional_str(payload.get("correlation_id")),
|
|
causation_id=_optional_str(payload.get("causation_id")),
|
|
actor=_actor_ref(payload.get("actor")),
|
|
tenant=_tenant_ref(payload.get("tenant")),
|
|
subject=_object_ref(payload.get("subject")),
|
|
resource=_object_ref(payload.get("resource")),
|
|
classification=cast(EventClassification, str(payload.get("classification") or "internal")),
|
|
institutional_context=_institutional_context(
|
|
payload.get("institutional_context")
|
|
),
|
|
)
|
|
|
|
|
|
def _institutional_context(
|
|
value: object,
|
|
) -> GovernedContextEnvelope | None:
|
|
return (
|
|
GovernedContextEnvelope.from_mapping(value)
|
|
if isinstance(value, Mapping)
|
|
else None
|
|
)
|
|
|
|
|
|
def _session(session: object) -> Session:
|
|
if not isinstance(session, Session):
|
|
raise TypeError("Audit outbox requires a SQLAlchemy Session")
|
|
return session
|
|
|
|
|
|
def _mapping(value: object) -> dict[str, Any]:
|
|
return dict(value) if isinstance(value, Mapping) else {}
|
|
|
|
|
|
def _optional_str(value: object) -> str | None:
|
|
return str(value) if value is not None else None
|
|
|
|
|
|
def _datetime(value: object) -> datetime:
|
|
if isinstance(value, datetime):
|
|
return value
|
|
if isinstance(value, str):
|
|
return datetime.fromisoformat(value)
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def _actor_ref(value: object) -> EventActorRef | None:
|
|
data = _mapping(value)
|
|
if not data:
|
|
return None
|
|
return EventActorRef(type=str(data["type"]), id=_optional_str(data.get("id")), label=_optional_str(data.get("label")))
|
|
|
|
|
|
def _tenant_ref(value: object) -> EventTenantRef | None:
|
|
data = _mapping(value)
|
|
if not data:
|
|
return None
|
|
return EventTenantRef(id=str(data["id"]), slug=_optional_str(data.get("slug")), label=_optional_str(data.get("label")))
|
|
|
|
|
|
def _object_ref(value: object) -> EventObjectRef | None:
|
|
data = _mapping(value)
|
|
if not data:
|
|
return None
|
|
return EventObjectRef(type=str(data["type"]), id=_optional_str(data.get("id")), label=_optional_str(data.get("label")))
|