diff --git a/README.md b/README.md index bfeca6e..d776460 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,14 @@ This repository owns the live `audit_log` table, audit API route contributions, and the target boundary for future audit sink/export capability work. +It also owns the audit command/event separation and production delivery +foundation: + +- `govoplan_audit.backend.commands` defines an in-process command bus for + requested work. +- `govoplan_audit.backend.outbox` persists governed platform events in + `audit_outbox_events` and dispatches pending rows with retry metadata. + See [docs/AUDIT_TRACE_CONTEXT.md](docs/AUDIT_TRACE_CONTEXT.md) for the standard operational context fields used by admin, installer, and module lifecycle audit entries. diff --git a/docs/AUDIT_TRACE_CONTEXT.md b/docs/AUDIT_TRACE_CONTEXT.md index 5da20eb..3b5bb3f 100644 --- a/docs/AUDIT_TRACE_CONTEXT.md +++ b/docs/AUDIT_TRACE_CONTEXT.md @@ -80,3 +80,26 @@ acceptance events should include: This keeps admin UI timelines, audit exports, and rollback diagnostics aligned without coupling modules to the audit table implementation. + +## Commands, Events, And Outbox Delivery + +Commands and events are separate concepts: + +- Commands are imperative requests to do work, for example `retention.run` or + `module.install`. They use `govoplan_audit.backend.commands.AuditCommand` + and `CommandBus`. +- Events are completed facts, for example `tenant.created` or + `retention_policy.run`. They use the core `PlatformEvent` envelope and may + be written to the audit outbox before delivery. + +`govoplan_audit.backend.outbox.SqlAuditOutbox` persists platform events in +`audit_outbox_events`. Dispatchers can later call +`dispatch_pending_platform_events()` to publish pending events and record retry +state. The outbox payload stores the full governed event envelope: +correlation/causation ids, actor, tenant, subject, resource, classification, +module id, event id, type, and payload. + +Application code should enqueue or publish facts only after the state change +they describe is known. Long-running operators and installers should model +requested work as commands first, then emit facts as events as each step +completes. diff --git a/pyproject.toml b/pyproject.toml index 4fc4e98..b4cec39 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,6 @@ requires-python = ">=3.12" authors = [{ name = "GovOPlaN" }] dependencies = [ "govoplan-core>=0.1.6", - "govoplan-access>=0.1.6", ] [tool.setuptools.packages.find] @@ -22,4 +21,3 @@ govoplan_audit = ["py.typed"] [project.entry-points."govoplan.modules"] audit = "govoplan_audit.backend.manifest:get_manifest" - diff --git a/src/govoplan_audit/backend/commands.py b/src/govoplan_audit/backend/commands.py new file mode 100644 index 0000000..1e1abec --- /dev/null +++ b/src/govoplan_audit/backend/commands.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +from collections import defaultdict +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any +import uuid + + +def new_command_id() -> str: + return str(uuid.uuid4()) + + +@dataclass(frozen=True, slots=True) +class AuditCommand: + type: str + module_id: str + payload: Mapping[str, Any] = field(default_factory=dict) + command_id: str = field(default_factory=new_command_id) + requested_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + requested_by: str | None = None + correlation_id: str | None = None + causation_id: str | None = None + + def to_dict(self) -> dict[str, Any]: + return { + "type": self.type, + "module_id": self.module_id, + "payload": dict(self.payload), + "command_id": self.command_id, + "requested_at": self.requested_at.isoformat(), + "requested_by": self.requested_by, + "correlation_id": self.correlation_id, + "causation_id": self.causation_id, + } + + +CommandHandler = Callable[[AuditCommand], None] + + +class CommandBus: + def __init__(self) -> None: + self._handlers: dict[str, list[CommandHandler]] = defaultdict(list) + + def subscribe(self, command_type: str, handler: CommandHandler) -> None: + self._handlers[command_type].append(handler) + + def dispatch(self, command: AuditCommand) -> None: + for handler in self._handlers.get(command.type, ()): + handler(command) + for handler in self._handlers.get("*", ()): + handler(command) diff --git a/src/govoplan_audit/backend/db/models.py b/src/govoplan_audit/backend/db/models.py index 5adb6d9..262f0e1 100644 --- a/src/govoplan_audit/backend/db/models.py +++ b/src/govoplan_audit/backend/db/models.py @@ -3,7 +3,9 @@ from __future__ import annotations import uuid from typing import Any -from sqlalchemy import ForeignKey, Index, JSON, String +from datetime import datetime + +from sqlalchemy import DateTime, ForeignKey, Index, Integer, JSON, String, Text, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column from govoplan_core.db.base import Base, TimestampMixin @@ -31,4 +33,28 @@ class AuditLog(Base, TimestampMixin): details: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) -__all__ = ["AuditLog", "new_uuid"] +class AuditOutboxEvent(Base, TimestampMixin): + __tablename__ = "audit_outbox_events" + __table_args__ = ( + UniqueConstraint("event_id", name="uq_audit_outbox_events_event_id"), + Index("ix_audit_outbox_events_status_next_attempt_at", "status", "next_attempt_at"), + Index("ix_audit_outbox_events_event_type", "event_type"), + Index("ix_audit_outbox_events_correlation_id", "correlation_id"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + event_id: Mapped[str] = mapped_column(String(36), nullable=False) + event_type: Mapped[str] = mapped_column(String(200), nullable=False) + module_id: Mapped[str] = mapped_column(String(100), nullable=False) + correlation_id: Mapped[str | None] = mapped_column(String(128), nullable=True) + causation_id: Mapped[str | None] = mapped_column(String(128), nullable=True) + classification: Mapped[str] = mapped_column(String(40), nullable=False, default="internal") + payload: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending", index=True) + attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + next_attempt_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + dispatched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + last_error: Mapped[str | None] = mapped_column(Text, nullable=True) + + +__all__ = ["AuditLog", "AuditOutboxEvent", "new_uuid"] diff --git a/src/govoplan_audit/backend/manifest.py b/src/govoplan_audit/backend/manifest.py index d17fb49..dcfc4c0 100644 --- a/src/govoplan_audit/backend/manifest.py +++ b/src/govoplan_audit/backend/manifest.py @@ -43,11 +43,11 @@ manifest = ModuleManifest( module_id="audit", metadata=Base.metadata, retirement_supported=True, - retirement_provider=drop_table_retirement_provider(audit_models.AuditLog, label="Audit"), + retirement_provider=drop_table_retirement_provider(audit_models.AuditLog, audit_models.AuditOutboxEvent, label="Audit"), retirement_notes="Destructive retirement drops audit-owned database tables after the installer captures a database snapshot.", ), uninstall_guard_providers=( - persistent_table_uninstall_guard(audit_models.AuditLog, label="Audit"), + persistent_table_uninstall_guard(audit_models.AuditLog, audit_models.AuditOutboxEvent, label="Audit"), ), capability_factories={ CAPABILITY_AUDIT_RECORDER: _audit_recorder, diff --git a/src/govoplan_audit/backend/outbox.py b/src/govoplan_audit/backend/outbox.py new file mode 100644 index 0000000..f32637e --- /dev/null +++ b/src/govoplan_audit/backend/outbox.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +from collections.abc import Callable, Mapping +from datetime import datetime, timedelta, timezone +from typing import Any, cast + +from sqlalchemy import or_ +from sqlalchemy.orm import Session + +from govoplan_audit.backend.db.models import AuditOutboxEvent +from govoplan_core.core.events import ( + EventActorRef, + EventClassification, + EventObjectRef, + EventTenantRef, + PlatformEvent, + ensure_event_trace, + publish_platform_event, +) + +EventDispatcher = Callable[[PlatformEvent], None] + + +class SqlAuditOutbox: + def enqueue(self, session: object, event: PlatformEvent) -> AuditOutboxEvent: + db = _session(session) + traced = ensure_event_trace(event) + 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) + db.flush() + return item + + def dispatch_pending( + self, + session: object, + *, + dispatcher: EventDispatcher = publish_platform_event, + limit: int = 100, + ) -> dict[str, int]: + db = _session(session) + now = datetime.now(timezone.utc) + rows = ( + db.query(AuditOutboxEvent) + .filter( + AuditOutboxEvent.status.in_(("pending", "failed")), + or_(AuditOutboxEvent.next_attempt_at.is_(None), AuditOutboxEvent.next_attempt_at <= now), + ) + .order_by(AuditOutboxEvent.created_at.asc(), AuditOutboxEvent.id.asc()) + .limit(max(1, min(int(limit), 500))) + .all() + ) + counts = {"selected": len(rows), "dispatched": 0, "failed": 0} + for row in rows: + try: + dispatcher(_event_from_payload(row.payload)) + except Exception as exc: # noqa: BLE001 - dispatcher errors must be retained for retry/diagnostics. + row.status = "failed" + row.attempts += 1 + row.last_error = str(exc) + row.next_attempt_at = now + _retry_delay(row.attempts) + counts["failed"] += 1 + continue + row.status = "dispatched" + row.attempts += 1 + row.dispatched_at = now + row.next_attempt_at = None + row.last_error = None + counts["dispatched"] += 1 + db.flush() + return counts + + +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 _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")), + ) + + +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"))) diff --git a/tests/test_audit_delivery.py b/tests/test_audit_delivery.py new file mode 100644 index 0000000..d859464 --- /dev/null +++ b/tests/test_audit_delivery.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from govoplan_audit.backend.commands import AuditCommand, CommandBus +from govoplan_audit.backend.db.models import AuditOutboxEvent +from govoplan_audit.backend.outbox import SqlAuditOutbox +from govoplan_core.core.events import EventActorRef, PlatformEvent +from govoplan_core.db.base import Base + + +class AuditCommandBusTests(unittest.TestCase): + def test_command_bus_dispatches_commands_separately_from_events(self) -> None: + bus = CommandBus() + seen: list[AuditCommand] = [] + wildcard: list[AuditCommand] = [] + + bus.subscribe("retention.run", seen.append) + bus.subscribe("*", wildcard.append) + command = AuditCommand(type="retention.run", module_id="policy", payload={"dry_run": True}) + + bus.dispatch(command) + + self.assertEqual([command], seen) + self.assertEqual([command], wildcard) + self.assertEqual("retention.run", command.to_dict()["type"]) + + +class AuditOutboxTests(unittest.TestCase): + def test_outbox_enqueues_governed_event_and_dispatches_pending_rows(self) -> None: + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(bind=engine, tables=[AuditOutboxEvent.__table__]) + Session = sessionmaker(bind=engine) + outbox = SqlAuditOutbox() + seen: list[PlatformEvent] = [] + + with Session() as session: + event = PlatformEvent( + type="tenant.created", + module_id="tenancy", + payload={"tenant_id": "tenant-1"}, + actor=EventActorRef(type="user", id="user-1"), + ) + row = outbox.enqueue(session, event) + + self.assertEqual("pending", row.status) + self.assertEqual("tenant.created", row.event_type) + self.assertEqual(event.event_id, row.event_id) + self.assertEqual(event.event_id, row.correlation_id) + self.assertEqual("user-1", row.payload["actor"]["id"]) + + counts = outbox.dispatch_pending(session, dispatcher=seen.append) + + self.assertEqual({"selected": 1, "dispatched": 1, "failed": 0}, counts) + self.assertEqual(1, len(seen)) + self.assertEqual("tenant.created", seen[0].type) + self.assertEqual("dispatched", row.status) + self.assertEqual(1, row.attempts) + self.assertIsNotNone(row.dispatched_at) + + def test_outbox_records_failed_dispatch_for_retry(self) -> None: + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(bind=engine, tables=[AuditOutboxEvent.__table__]) + Session = sessionmaker(bind=engine) + outbox = SqlAuditOutbox() + + with Session() as session: + row = outbox.enqueue(session, PlatformEvent(type="demo.failed", module_id="audit")) + + counts = outbox.dispatch_pending(session, dispatcher=lambda event: (_ for _ in ()).throw(RuntimeError("offline"))) + + self.assertEqual({"selected": 1, "dispatched": 0, "failed": 1}, counts) + self.assertEqual("failed", row.status) + self.assertEqual(1, row.attempts) + self.assertEqual("offline", row.last_error) + self.assertIsNotNone(row.next_attempt_at) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_audit_module_contract.py b/tests/test_audit_module_contract.py new file mode 100644 index 0000000..5f082bc --- /dev/null +++ b/tests/test_audit_module_contract.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +import pathlib +import tomllib +import unittest + + +ROOT = pathlib.Path(__file__).resolve().parents[1] + + +class AuditModuleContractTests(unittest.TestCase): + def test_audit_package_does_not_hard_require_access(self) -> None: + project = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))["project"] + dependencies = tuple(project["dependencies"]) + + self.assertIn("govoplan-core>=0.1.6", dependencies) + self.assertFalse(any(item.startswith("govoplan-access") for item in dependencies)) + + def test_audit_source_does_not_import_access_implementation(self) -> None: + offenders: list[str] = [] + for path in (ROOT / "src" / "govoplan_audit").rglob("*.py"): + source = path.read_text(encoding="utf-8") + if "govoplan_access" in source: + offenders.append(str(path.relative_to(ROOT))) + + self.assertEqual([], offenders) + + +if __name__ == "__main__": + unittest.main()