22 Commits
Author SHA1 Message Date
zemion 1d993b26c8 Release v0.1.17
Module Package Release / publish-packages (push) Successful in 12s
2026-08-05 20:33:57 +02:00
zemion 4613b54c06 Release v0.1.16
Module Package Release / publish-packages (push) Successful in 12s
2026-08-05 19:51:58 +02:00
zemion 6f095eb562 Release v0.1.15
Module Package Release / publish-packages (push) Successful in 10s
2026-08-04 15:18:08 +02:00
zemion 0cbb249d74 Make package publication retries hash-safe 2026-08-04 14:32:18 +02:00
zemion f0d2916d46 Harden module package publication 2026-08-04 14:02:39 +02:00
zemion c1e6c15866 Partition durable events by tenant entitlement 2026-08-04 09:29:35 +02:00
zemion 451361cc05 Add protected package release workflow 2026-08-04 04:14:02 +02:00
zemion 6d3fcc1572 Migrate Audit evidence interfaces 2026-08-03 10:43:24 +02:00
zemion 4177287b22 feat: strengthen transactional audit delivery 2026-08-01 17:48:23 +02:00
zemion 57ceef0173 Align WebUI runtime peer dependencies 2026-07-31 02:48:56 +02:00
zemion 078e9144b1 feat: classify audit administration surfaces 2026-07-30 17:42:06 +02:00
zemion f74e8cf85b Persist durable event consumer delivery state 2026-07-29 17:34:52 +02:00
zemion 0167ab752a feat: publish durable audit outbox events 2026-07-29 14:16:28 +02:00
zemion 1479946729 Declare administration View surfaces 2026-07-28 21:04:54 +02:00
zemion 130f738970 fix(webui): require Core 0.1.9 for table actions 2026-07-21 13:46:20 +02:00
zemion a34935da02 refactor(api): share full audit delta response 2026-07-21 13:29:19 +02:00
zemion 86b20c65cb refactor(webui): use central table actions 2026-07-21 12:01:24 +02:00
zemion a96dc228b8 Clean audit security scan and test resources 2026-07-21 03:16:23 +02:00
zemion 5e4f84a789 intermittent commit 2026-07-14 13:22:10 +02:00
zemion d3d2c60d7d Release v0.1.8 2026-07-11 16:49:00 +02:00
zemion 7ffb16d981 Release v0.1.7 2026-07-11 02:34:56 +02:00
zemion ae70cac70f Add audit command and outbox foundations 2026-07-11 00:46:09 +02:00
27 changed files with 2823 additions and 165 deletions
+270
View File
@@ -0,0 +1,270 @@
name: Module Package Release
on:
push:
tags:
- "v*"
workflow_dispatch:
inputs:
release_tag:
description: Existing protected version tag to publish
required: true
type: string
jobs:
publish-packages:
runs-on: ubuntu-latest
env:
GITEA_REPOSITORY: ${{ gitea.repository }}
steps:
- uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5
with:
fetch-depth: 0
- uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065
with:
python-version: "3.12"
- uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020
with:
node-version: "22"
- name: Select and validate protected release tag
shell: bash
env:
REQUESTED_TAG: ${{ inputs.release_tag }}
TRIGGER_TAG: ${{ gitea.ref_name }}
run: |
set -euo pipefail
tag="${REQUESTED_TAG:-$TRIGGER_TAG}"
case "$tag" in
v[0-9]*.[0-9]*.[0-9]*) ;;
*) echo "Release tag must start with a SemVer-shaped vX.Y.Z value" >&2; exit 1 ;;
esac
git fetch --force origin "refs/tags/$tag:refs/tags/$tag" refs/heads/main:refs/remotes/origin/main
tag_commit="$(git rev-list -n 1 "$tag")"
git merge-base --is-ancestor "$tag_commit" refs/remotes/origin/main || {
echo "Release tag is not contained in main" >&2
exit 1
}
git checkout --detach "$tag"
printf 'RELEASE_TAG=%s\n' "$tag" >> "$GITEA_ENV"
printf 'SOURCE_DATE_EPOCH=%s\n' "$(git show -s --format=%ct HEAD)" >> "$GITEA_ENV"
- name: Validate package versions
run: |
python - <<'PY'
import json
from pathlib import Path
import os
import re
import tomllib
tag = os.environ["RELEASE_TAG"]
expected = tag.removeprefix("v")
project = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))["project"]
if project.get("version") != expected:
raise SystemExit(f"pyproject version {project.get('version')!r} does not match {tag}")
if re.fullmatch(r"govoplan-[a-z0-9-]+", str(project.get("name", ""))) is None:
raise SystemExit("Python distribution name must use the govoplan-* namespace")
webui = Path("webui/package.json")
if webui.is_file():
package = json.loads(webui.read_text(encoding="utf-8"))
if package.get("version") != expected:
raise SystemExit(f"WebUI version {package.get('version')!r} does not match {tag}")
if re.fullmatch(r"@govoplan/[a-z0-9-]+-webui", str(package.get("name", ""))) is None:
raise SystemExit("WebUI package name must use the @govoplan/*-webui namespace")
release = Path("webui/package.release.json")
if release.is_file():
release_package = json.loads(release.read_text(encoding="utf-8"))
if (
release_package.get("name") != package.get("name")
or release_package.get("version") != expected
):
raise SystemExit("WebUI release package identity does not match package.json and the release tag")
PY
- name: Build immutable package artifacts
shell: bash
run: |
set -euo pipefail
python -m pip install --disable-pip-version-check build==1.5.0 twine==7.0.0
rm -rf dist .package-webui
python -m build --wheel --outdir dist
python -m twine check dist/*.whl
if [[ -f webui/package.json ]]; then
mkdir .package-webui
cp -a webui/. .package-webui/
rm -rf .package-webui/node_modules .package-webui/dist
if [[ -f .package-webui/package.release.json ]]; then
cp .package-webui/package.release.json .package-webui/package.json
fi
node <<'NODE'
const fs = require("node:fs");
const path = ".package-webui/package.json";
const packageJson = JSON.parse(fs.readFileSync(path, "utf8"));
const groups = ["dependencies", "optionalDependencies", "peerDependencies"];
for (const group of groups) {
for (const [name, specifier] of Object.entries(packageJson[group] || {})) {
if (!name.startsWith("@govoplan/")) continue;
if (typeof specifier !== "string") {
throw new Error(`${group}.${name} must use a string version`);
}
const packageSlug = name.slice("@govoplan/".length);
if (!packageSlug.endsWith("-webui")) {
throw new Error(`${group}.${name} is outside the WebUI package namespace`);
}
const repository = `govoplan-${packageSlug.slice(0, -"-webui".length)}`;
const escapedRepository = repository.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
const gitTag = specifier.match(
new RegExp(
`^git\\+(?:ssh://git@|https://)git\\.add-ideas\\.de/(?:GovOPlaN|add-ideas)/${escapedRepository}\\.git#v([0-9]+\\.[0-9]+\\.[0-9]+)$`,
),
);
if (gitTag) {
packageJson[group][name] = gitTag[1];
continue;
}
if (specifier.startsWith("file:") || specifier.startsWith("git+")) {
throw new Error(
`${group}.${name} must resolve to an exact registry version for publication`,
);
}
}
}
delete packageJson.private;
fs.writeFileSync(path, `${JSON.stringify(packageJson, null, 2)}\n`);
NODE
npm pkg delete private --prefix .package-webui
(cd .package-webui && npm pack --ignore-scripts --pack-destination ../dist)
fi
python - <<'PY'
import hashlib
import json
from pathlib import Path
import os
import subprocess
artifacts = []
for path in sorted(Path("dist").iterdir()):
if path.suffix not in {".whl", ".tgz"}:
continue
digest = hashlib.sha256(path.read_bytes()).hexdigest()
artifacts.append({"filename": path.name, "sha256": digest, "size": path.stat().st_size})
payload = {
"schema_version": "1",
"repository": os.environ["GITEA_REPOSITORY"],
"tag": os.environ["RELEASE_TAG"],
"commit": subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip(),
"artifacts": artifacts,
}
Path("dist/package-artifacts.json").write_text(
json.dumps(payload, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
PY
- name: Retain package hash evidence
uses: actions/upload-artifact@a8a3f3ad30e3422c9c7b888a15615d19a852ae32
with:
name: module-packages-${{ gitea.ref_name }}
path: dist/package-artifacts.json
- name: Check immutable registry state
shell: bash
env:
PACKAGE_TOKEN: ${{ secrets.GOVOPLAN_PACKAGE_TOKEN }}
run: |
set -euo pipefail
test -n "$PACKAGE_TOKEN"
python - <<'PY'
import hashlib
import json
import os
from pathlib import Path
import tomllib
from urllib.error import HTTPError
from urllib.parse import quote
from urllib.request import Request, urlopen
api_root = "https://git.add-ideas.de/api/v1/packages/GovOPlaN"
token = os.environ["PACKAGE_TOKEN"]
def should_publish(kind, name, version, path):
package_url = "/".join(
(api_root, kind, quote(name, safe=""), quote(version, safe=""), "files")
)
request = Request(
package_url,
headers={"Accept": "application/json", "Authorization": f"token {token}"},
)
try:
with urlopen(request, timeout=30) as response:
files = json.load(response)
except HTTPError as exc:
if exc.code == 404:
print(f"{kind} package {name}=={version} is not published yet")
return True
raise
if not isinstance(files, list) or len(files) != 1:
raise SystemExit(
f"immutable {kind} package {name}=={version} has an unexpected file set"
)
expected_sha256 = hashlib.sha256(path.read_bytes()).hexdigest()
if files[0].get("sha256") != expected_sha256:
raise SystemExit(
f"immutable {kind} package {name}=={version} already exists with a different SHA-256"
)
print(f"verified existing {kind} package {name}=={version} ({expected_sha256})")
return False
project = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))["project"]
wheels = tuple(Path("dist").glob("*.whl"))
if len(wheels) != 1:
raise SystemExit("release build must contain exactly one wheel")
publish_pypi = should_publish(
"pypi", str(project["name"]), str(project["version"]), wheels[0]
)
tarballs = tuple(Path("dist").glob("*.tgz"))
if len(tarballs) > 1:
raise SystemExit("release build must contain at most one npm package")
publish_npm = False
if tarballs:
webui = json.loads(
Path(".package-webui/package.json").read_text(encoding="utf-8")
)
publish_npm = should_publish(
"npm", str(webui["name"]), str(webui["version"]), tarballs[0]
)
with Path(os.environ["GITEA_ENV"]).open("a", encoding="utf-8") as env_file:
env_file.write(f"PUBLISH_PYPI={int(publish_pypi)}\n")
env_file.write(f"PUBLISH_NPM={int(publish_npm)}\n")
PY
- name: Publish wheel and WebUI package
shell: bash
env:
PACKAGE_USERNAME: ${{ secrets.GOVOPLAN_PACKAGE_USERNAME }}
PACKAGE_TOKEN: ${{ secrets.GOVOPLAN_PACKAGE_TOKEN }}
run: |
set -euo pipefail
test -n "$PACKAGE_USERNAME"
test -n "$PACKAGE_TOKEN"
if [[ "$PUBLISH_PYPI" == 1 ]]; then
TWINE_USERNAME="$PACKAGE_USERNAME" TWINE_PASSWORD="$PACKAGE_TOKEN" \
python -m twine upload --non-interactive \
--repository-url https://git.add-ideas.de/api/packages/GovOPlaN/pypi \
dist/*.whl
else
echo "Exact wheel is already present; skipping immutable retry."
fi
shopt -s nullglob
webui_packages=(dist/*.tgz)
if (( ${#webui_packages[@]} )) && [[ "$PUBLISH_NPM" == 1 ]]; then
npmrc="$(mktemp)"
trap 'rm -f "$npmrc"' EXIT
chmod 600 "$npmrc"
printf '%s\n' \
'@govoplan:registry=https://git.add-ideas.de/api/packages/GovOPlaN/npm/' \
"//git.add-ideas.de/api/packages/GovOPlaN/npm/:_authToken=$PACKAGE_TOKEN" \
> "$npmrc"
NPM_CONFIG_USERCONFIG="$npmrc" npm publish "./${webui_packages[0]}" \
--ignore-scripts --access public \
--registry https://git.add-ideas.de/api/packages/GovOPlaN/npm/
elif (( ${#webui_packages[@]} )); then
echo "Exact WebUI package is already present; skipping immutable retry."
fi
+16
View File
@@ -0,0 +1,16 @@
# GovOPlaN Audit Codex Guide
## Scope
This repository owns durable audit records, audit administration surfaces, retention behavior, and the transactional platform-event outbox.
## Documentation Contract
- Treat documentation as part of every behavior change. Update this module's manifest-driven `DocumentationTopic` contributions for affected user and administrator behavior.
- Keep feature content here; `govoplan-docs` projects it without importing Audit internals.
- Maintain a static user/admin baseline and run `/mnt/DATA/git/govoplan/tools/checks/check-manifest-shapes.py` after behavior or manifest changes.
## Boundaries
- Store bounded evidence and trace context, not arbitrary feature payloads.
- Preserve transactional recording, retention, redaction, and retry guarantees.
+24 -4
View File
@@ -1,12 +1,32 @@
# GovOPlaN Audit
`govoplan-audit` owns audit API route contributions during the GovOPlaN module
split.
<!-- govoplan-repository-type:start -->
**Repository type:** module (platform).
<!-- govoplan-repository-type:end -->
`govoplan-audit` owns audit API route contributions and audit administration
WebUI sections during the GovOPlaN module split.
This repository owns the live `audit_log` table, audit API route
contributions, and the target boundary for future audit sink/export capability
work.
contributions, the `@govoplan/audit-webui` package, and the target boundary
for future audit sink/export capability work.
The WebUI package contributes the `system-audit` and `tenant-audit` admin
sections through the shared `admin.sections` UI capability. The admin shell
does not render audit panels unless this module is installed and enabled.
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.
The administration surface archetypes, consequence classification, and
verification contract are recorded in
[docs/INTERFACE_PATTERN_MIGRATION.md](docs/INTERFACE_PATTERN_MIGRATION.md).
+35
View File
@@ -80,3 +80,38 @@ 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` and one durable state row per allowlisted consumer in
`audit_outbox_deliveries`. Dispatchers supply stable consumer IDs and
idempotent handlers. Consumer work and its delivered marker share one database
transaction; retries reuse the stable `<event-id>:<consumer-id>` delivery key.
Bounded failures are quarantined instead of retried forever. The outbox payload
stores the full governed event envelope:
correlation/causation ids, actor, tenant, subject, resource, classification,
module id, event id, type, and payload.
Public and internal events may use an allowlisted subscription directly.
Confidential and restricted subscriptions additionally require a persisted
policy-decision reference. Operators can inspect delivery metrics at
`GET /api/v1/admin/audit/event-delivery/metrics` and replay a retrying or
quarantined delivery with a reason through
`POST /api/v1/admin/audit/event-deliveries/{event_id}/{consumer_id}/replay`.
Replay itself is written to the audit log. Successful envelopes are subject to
configured retention; quarantined evidence is not removed automatically.
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.
+40
View File
@@ -0,0 +1,40 @@
# Audit Interface Pattern Migration
Audit contributes two read-only administration surfaces through the shared
`admin.sections` capability. Both use the platform's monitoring and evidence
archetype.
## Surface Map
| Surface | Authority | Pattern | Consequence class |
| --- | --- | --- | --- |
| `audit.admin.system` | `system:audit:read` | Server-filtered evidence grid and event inspector | Read-only evidence disclosure |
| `audit.admin.tenant` | `audit:read` for the active tenant | Server-filtered evidence grid and event inspector | Read-only evidence disclosure |
Audit does not expose mutation or destructive actions in these panels. The
only row action opens an inspection dialog; reload preserves the stable shell
and existing evidence while a newer projection is requested.
## Interaction Contract
- Core owns the admin layout, DataGrid, dialog, action group, loading/error
treatment, disabled-action explanation, and documentation link.
- Filtering, sorting, counts, and paging are server-owned. The first page may
apply bounded delta updates using an opaque watermark; a full response
remains authoritative when the delta contract cannot be used.
- System and tenant panels remain distinct and are registered only with their
respective read scopes. Tenant selection is never accepted as a free-form
client override.
- The event inspector renders stable actor, action, object, tenant, timestamp,
and structured detail rows. It does not add editing, replay, export, or raw
credential access.
- Contextual help resolves through `audit.read-authorized-evidence`; operational
recording, retention, and outbox guidance remains in the separate admin
topic.
## Verification
Run the Audit backend suite and `npm run test:interface-patterns` in `webui`.
The structural test guards shared components, localized labels, contextual
help, server paging, readable detail projection, and absence of private sibling
imports or browser-native dialogs.
+32
View File
@@ -0,0 +1,32 @@
{
"name": "@govoplan/audit-webui",
"version": "0.1.17",
"private": true,
"type": "module",
"main": "webui/src/index.ts",
"module": "webui/src/index.ts",
"types": "webui/src/index.ts",
"exports": {
".": {
"types": "./webui/src/index.ts",
"import": "./webui/src/index.ts"
}
},
"files": [
"webui/src",
"README.md",
"LICENSE"
],
"peerDependencies": {
"@govoplan/core-webui": "^0.1.17",
"lucide-react": "^1.23.0",
"react": ">=19.2.7 <20",
"react-dom": ">=19.2.7 <20",
"react-router": ">=8.3.0 <9"
},
"peerDependenciesMeta": {
"@govoplan/core-webui": {
"optional": true
}
}
}
+2 -4
View File
@@ -4,14 +4,13 @@ build-backend = "setuptools.build_meta"
[project]
name = "govoplan-audit"
version = "0.1.6"
version = "0.1.17"
description = "GovOPlaN audit platform module."
readme = "README.md"
requires-python = ">=3.12"
authors = [{ name = "GovOPlaN" }]
dependencies = [
"govoplan-core>=0.1.6",
"govoplan-access>=0.1.6",
"govoplan-core>=0.1.17",
]
[tool.setuptools.packages.find]
@@ -22,4 +21,3 @@ govoplan_audit = ["py.typed"]
[project.entry-points."govoplan.modules"]
audit = "govoplan_audit.backend.manifest:get_manifest"
+266 -150
View File
@@ -1,6 +1,8 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import and_, false, func, or_
@@ -8,7 +10,12 @@ from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal, has_scope, require_any_scope
from govoplan_audit.backend.db.models import AuditLog
from govoplan_core.audit.logging import AUDIT_MODULE_ID, AUDIT_SYSTEM_EVENTS_COLLECTION, AUDIT_TENANT_EVENTS_COLLECTION
from govoplan_core.audit.logging import (
AUDIT_MODULE_ID,
AUDIT_SYSTEM_EVENTS_COLLECTION,
AUDIT_TENANT_EVENTS_COLLECTION,
audit_from_principal,
)
from govoplan_core.core.access import CAPABILITY_ACCESS_ADMINISTRATION, AccessAdministration
from govoplan_core.core.change_sequence import decode_sequence_watermark, encode_sequence_watermark, max_sequence_id, sequence_entries_since, sequence_watermark_is_expired
from govoplan_core.core.pagination import KeysetCursorError, decode_keyset_cursor, encode_keyset_cursor, keyset_query_fingerprint
@@ -16,13 +23,38 @@ from govoplan_core.core.runtime import get_registry
from govoplan_core.db.session import get_session
from govoplan_core.tenancy.scope import Tenant
from .schemas import AuditAdminDeltaResponse, AuditAdminItem, AuditAdminListResponse, AuditLogItemResponse, AuditLogListResponse
from govoplan_core.core.events import platform_event_outbox
from .schemas import (
AuditAdminDeltaResponse,
AuditAdminItem,
AuditAdminListResponse,
AuditLogItemResponse,
AuditLogListResponse,
EventDeliveryMetricsResponse,
EventDeliveryReplayRequest,
EventDeliveryReplayResponse,
)
router = APIRouter(tags=["audit"])
AUDIT_ADMIN_CURSOR_SCOPE = "audit.admin"
@dataclass(slots=True)
class AuditAdminQueryContext:
query: Any
access_admin: AccessAdministration
effective_scope: str
resolved_tenant_id: str | None
sort_column: Any
order: Any
total: int
effective_page_size: int
pages: int
fingerprint: str
def _access_administration() -> AccessAdministration:
registry = get_registry()
if registry is None or not registry.has_capability(CAPABILITY_ACCESS_ADMINISTRATION):
@@ -180,6 +212,36 @@ def _audit_delta_response_watermark(
return encode_sequence_watermark(entries[-1].id) if has_more and entries else _audit_delta_watermark(session, effective_scope=effective_scope, tenant_id=tenant_id)
def _full_audit_delta_response(
session: Session,
*,
context: AuditAdminQueryContext,
page_query: Any,
start_cursor: str | None,
sort_by: str,
sort_direction: str,
) -> AuditAdminDeltaResponse:
rows_plus_one = page_query.order_by(context.order, AuditLog.id.desc()).limit(context.effective_page_size + 1).all()
rows = rows_plus_one[:context.effective_page_size]
next_cursor = (
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=context.fingerprint)
if len(rows_plus_one) > context.effective_page_size and rows else None
)
return AuditAdminDeltaResponse(
total=context.total,
page=1,
page_size=context.effective_page_size,
pages=context.pages,
cursor=start_cursor,
next_cursor=next_cursor,
items=_audit_items(session, rows, context.access_admin),
deleted=[],
watermark=_audit_delta_watermark(session, effective_scope=context.effective_scope, tenant_id=context.resolved_tenant_id),
has_more=False,
full=True,
)
def _audit_items(session: Session, rows: list[AuditLog], access_admin: AccessAdministration) -> list[AuditAdminItem]:
actor_email_by_user_id = access_admin.actor_email_by_user_id(session, {row.user_id for row in rows if row.user_id})
return [
@@ -286,26 +348,23 @@ def _audit_cursor_condition(sort_column, *, sort_by: str, sort_direction: str, c
return or_(primary_after, and_(sort_column == sort_value, AuditLog.id < cursor_id))
@router.get("/admin/audit", response_model=AuditAdminListResponse)
def list_admin_audit(
tenant_id: str | None = Query(default=None),
all_tenants: bool = Query(default=False),
audit_scope: str | None = Query(default=None, alias="scope"),
limit: int = Query(default=100, ge=1, le=500),
offset: int = Query(default=0, ge=0),
page: int | None = Query(default=None, ge=1),
page_size: int | None = Query(default=None, ge=1, le=500),
cursor: str | None = Query(default=None),
sort_by: str = Query(default="time"),
sort_direction: str = Query(default="desc"),
filter_time: str | None = Query(default=None),
filter_actor: str | None = Query(default=None),
filter_action: str | None = Query(default=None),
filter_object: str | None = Query(default=None),
filter_tenant: str | None = Query(default=None),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope("audit:read", "system:audit:read")),
):
def _prepare_audit_admin_query(
session: Session,
principal: ApiPrincipal,
*,
tenant_id: str | None,
all_tenants: bool,
audit_scope: str | None,
limit: int,
page_size: int | None,
sort_by: str,
sort_direction: str,
filter_time: str | None,
filter_actor: str | None,
filter_action: str | None,
filter_object: str | None,
filter_tenant: str | None,
) -> AuditAdminQueryContext:
effective_scope = audit_scope or ("all" if all_tenants else "tenant")
if effective_scope not in {"tenant", "system", "all"}:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Audit scope must be tenant, system or all.")
@@ -333,14 +392,13 @@ def list_admin_audit(
object_text = func.coalesce(AuditLog.object_type, "") + " " + func.coalesce(AuditLog.object_id, "")
access_admin = _access_administration()
filters = [
for condition in (
_audit_time_filter(filter_time),
_audit_actor_filter(access_admin, session, filter_actor),
_audit_text_filter(AuditLog.action, filter_action),
_audit_text_filter(object_text, filter_object),
_audit_text_filter(AuditLog.tenant_id, filter_tenant),
]
for condition in filters:
):
if condition is not None:
query = query.filter(condition)
@@ -353,7 +411,6 @@ def list_admin_audit(
}
sort_column = sort_columns[sort_by]
order = sort_column.asc() if sort_direction == "asc" else sort_column.desc()
ordered_query = query.order_by(order, AuditLog.id.desc())
total = query.count()
effective_page_size = page_size or limit
pages = max(1, (total + effective_page_size - 1) // effective_page_size)
@@ -372,47 +429,100 @@ def list_admin_audit(
sort_direction=sort_direction,
filters=filters,
)
return AuditAdminQueryContext(
query=query,
access_admin=access_admin,
effective_scope=effective_scope,
resolved_tenant_id=resolved_tenant_id,
sort_column=sort_column,
order=order,
total=total,
effective_page_size=effective_page_size,
pages=pages,
fingerprint=fingerprint,
)
@router.get("/admin/audit", response_model=AuditAdminListResponse)
def list_admin_audit(
tenant_id: str | None = Query(default=None),
all_tenants: bool = Query(default=False),
audit_scope: str | None = Query(default=None, alias="scope"),
limit: int = Query(default=100, ge=1, le=500),
offset: int = Query(default=0, ge=0),
page: int | None = Query(default=None, ge=1),
page_size: int | None = Query(default=None, ge=1, le=500),
cursor: str | None = Query(default=None),
sort_by: str = Query(default="time"),
sort_direction: str = Query(default="desc"),
filter_time: str | None = Query(default=None),
filter_actor: str | None = Query(default=None),
filter_action: str | None = Query(default=None),
filter_object: str | None = Query(default=None),
filter_tenant: str | None = Query(default=None),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope("audit:read", "system:audit:read")),
):
context = _prepare_audit_admin_query(
session,
principal,
tenant_id=tenant_id,
all_tenants=all_tenants,
audit_scope=audit_scope,
limit=limit,
page_size=page_size,
sort_by=sort_by,
sort_direction=sort_direction,
filter_time=filter_time,
filter_actor=filter_actor,
filter_action=filter_action,
filter_object=filter_object,
filter_tenant=filter_tenant,
)
ordered_query = context.query.order_by(context.order, AuditLog.id.desc())
start_cursor: str | None = None
if cursor:
try:
cursor_values = decode_keyset_cursor(AUDIT_ADMIN_CURSOR_SCOPE, cursor, fingerprint=fingerprint)
cursor_values = decode_keyset_cursor(AUDIT_ADMIN_CURSOR_SCOPE, cursor, fingerprint=context.fingerprint)
if cursor_values is None:
raise KeysetCursorError("Invalid pagination cursor")
page_query = query.filter(_audit_cursor_condition(sort_column, sort_by=sort_by, sort_direction=sort_direction, cursor_values=cursor_values))
page_query = context.query.filter(
_audit_cursor_condition(context.sort_column, sort_by=sort_by, sort_direction=sort_direction, cursor_values=cursor_values)
)
except KeysetCursorError as exc:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
effective_page = page or (offset // effective_page_size + 1)
effective_page = page or (offset // context.effective_page_size + 1)
effective_offset = 0
start_cursor = cursor
else:
if page is not None or page_size is not None:
effective_page = min(page or 1, pages)
effective_offset = (effective_page - 1) * effective_page_size
effective_page = min(page or 1, context.pages)
effective_offset = (effective_page - 1) * context.effective_page_size
else:
effective_page = offset // effective_page_size + 1
effective_page = offset // context.effective_page_size + 1
effective_offset = offset
page_query = query
page_query = context.query
if effective_offset > 0:
previous_row = ordered_query.offset(effective_offset - 1).limit(1).first()
if previous_row is not None:
start_cursor = _audit_cursor_for_row(previous_row, sort_by=sort_by, sort_direction=sort_direction, fingerprint=fingerprint)
start_cursor = _audit_cursor_for_row(previous_row, sort_by=sort_by, sort_direction=sort_direction, fingerprint=context.fingerprint)
rows_plus_one = page_query.order_by(order, AuditLog.id.desc()).offset(effective_offset).limit(effective_page_size + 1).all()
rows = rows_plus_one[:effective_page_size]
rows_plus_one = page_query.order_by(context.order, AuditLog.id.desc()).offset(effective_offset).limit(context.effective_page_size + 1).all()
rows = rows_plus_one[:context.effective_page_size]
next_cursor = (
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=fingerprint)
if len(rows_plus_one) > effective_page_size and rows else None
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=context.fingerprint)
if len(rows_plus_one) > context.effective_page_size and rows else None
)
return AuditAdminListResponse(
total=total,
total=context.total,
page=effective_page,
page_size=effective_page_size,
pages=pages,
page_size=context.effective_page_size,
pages=context.pages,
cursor=start_cursor,
next_cursor=next_cursor,
items=_audit_items(session, rows, access_admin),
items=_audit_items(session, rows, context.access_admin),
)
@@ -435,150 +545,81 @@ def list_admin_audit_delta(
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope("audit:read", "system:audit:read")),
):
effective_scope = audit_scope or ("all" if all_tenants else "tenant")
if effective_scope not in {"tenant", "system", "all"}:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Audit scope must be tenant, system or all.")
if sort_by not in {"time", "actor", "action", "object", "tenant"}:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Unsupported audit sort column.")
if sort_direction not in {"asc", "desc"}:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Audit sort direction must be asc or desc.")
query = session.query(AuditLog)
resolved_tenant_id: str | None = None
if effective_scope != "all":
query = query.filter(AuditLog.scope == effective_scope)
if effective_scope == "system":
if not has_scope(principal, "system:audit:read"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: system:audit:read")
elif effective_scope == "all" or all_tenants:
if not has_scope(principal, "system:audit:read"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: system:audit:read")
else:
if not has_scope(principal, "audit:read"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: audit:read")
tenant = _resolve_tenant(session, principal, tenant_id)
resolved_tenant_id = tenant.id
query = query.filter(AuditLog.tenant_id == tenant.id)
object_text = func.coalesce(AuditLog.object_type, "") + " " + func.coalesce(AuditLog.object_id, "")
access_admin = _access_administration()
filters = [
_audit_time_filter(filter_time),
_audit_actor_filter(access_admin, session, filter_actor),
_audit_text_filter(AuditLog.action, filter_action),
_audit_text_filter(object_text, filter_object),
_audit_text_filter(AuditLog.tenant_id, filter_tenant),
]
for condition in filters:
if condition is not None:
query = query.filter(condition)
sort_columns = {
"time": AuditLog.created_at,
"actor": func.coalesce(AuditLog.user_id, "System"),
"action": AuditLog.action,
"object": object_text,
"tenant": func.coalesce(AuditLog.tenant_id, ""),
}
sort_column = sort_columns[sort_by]
order = sort_column.asc() if sort_direction == "asc" else sort_column.desc()
total = query.count()
effective_page_size = page_size or limit
pages = max(1, (total + effective_page_size - 1) // effective_page_size)
filters = _audit_filter_params(
context = _prepare_audit_admin_query(
session,
principal,
tenant_id=tenant_id,
all_tenants=all_tenants,
audit_scope=audit_scope,
limit=limit,
page_size=page_size,
sort_by=sort_by,
sort_direction=sort_direction,
filter_time=filter_time,
filter_actor=filter_actor,
filter_action=filter_action,
filter_object=filter_object,
filter_tenant=filter_tenant,
)
fingerprint = _audit_cursor_fingerprint(
effective_scope=effective_scope,
tenant_id=resolved_tenant_id,
page_size=effective_page_size,
sort_by=sort_by,
sort_direction=sort_direction,
filters=filters,
)
start_cursor: str | None = None
page_query = query
page_query = context.query
if cursor:
try:
cursor_values = decode_keyset_cursor(AUDIT_ADMIN_CURSOR_SCOPE, cursor, fingerprint=fingerprint)
cursor_values = decode_keyset_cursor(AUDIT_ADMIN_CURSOR_SCOPE, cursor, fingerprint=context.fingerprint)
if cursor_values is None:
raise KeysetCursorError("Invalid pagination cursor")
page_query = query.filter(_audit_cursor_condition(sort_column, sort_by=sort_by, sort_direction=sort_direction, cursor_values=cursor_values))
page_query = context.query.filter(
_audit_cursor_condition(context.sort_column, sort_by=sort_by, sort_direction=sort_direction, cursor_values=cursor_values)
)
start_cursor = cursor
except KeysetCursorError as exc:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
if since is None:
rows_plus_one = page_query.order_by(order, AuditLog.id.desc()).limit(effective_page_size + 1).all()
rows = rows_plus_one[:effective_page_size]
next_cursor = (
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=fingerprint)
if len(rows_plus_one) > effective_page_size and rows else None
)
return AuditAdminDeltaResponse(
total=total,
page=1,
page_size=effective_page_size,
pages=pages,
cursor=start_cursor,
next_cursor=next_cursor,
items=_audit_items(session, rows, access_admin),
deleted=[],
watermark=_audit_delta_watermark(session, effective_scope=effective_scope, tenant_id=resolved_tenant_id),
has_more=False,
full=True,
return _full_audit_delta_response(
session,
context=context,
page_query=page_query,
start_cursor=start_cursor,
sort_by=sort_by,
sort_direction=sort_direction,
)
entries, has_more = _audit_delta_entries(
session,
effective_scope=effective_scope,
tenant_id=resolved_tenant_id,
effective_scope=context.effective_scope,
tenant_id=context.resolved_tenant_id,
since=since,
limit=effective_page_size,
limit=context.effective_page_size,
)
if entries is None:
rows_plus_one = page_query.order_by(order, AuditLog.id.desc()).limit(effective_page_size + 1).all()
rows = rows_plus_one[:effective_page_size]
next_cursor = (
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=fingerprint)
if len(rows_plus_one) > effective_page_size and rows else None
)
return AuditAdminDeltaResponse(
total=total,
page=1,
page_size=effective_page_size,
pages=pages,
cursor=start_cursor,
next_cursor=next_cursor,
items=_audit_items(session, rows, access_admin),
deleted=[],
watermark=_audit_delta_watermark(session, effective_scope=effective_scope, tenant_id=resolved_tenant_id),
has_more=False,
full=True,
return _full_audit_delta_response(
session,
context=context,
page_query=page_query,
start_cursor=start_cursor,
sort_by=sort_by,
sort_direction=sort_direction,
)
changed_ids = [entry.resource_id for entry in entries if entry.resource_type == "audit_log"]
rows = (
page_query.filter(AuditLog.id.in_(changed_ids)).order_by(order, AuditLog.id.desc()).limit(effective_page_size).all()
page_query.filter(AuditLog.id.in_(changed_ids)).order_by(context.order, AuditLog.id.desc()).limit(context.effective_page_size).all()
if changed_ids else []
)
return AuditAdminDeltaResponse(
total=total,
total=context.total,
page=1,
page_size=effective_page_size,
pages=pages,
page_size=context.effective_page_size,
pages=context.pages,
cursor=start_cursor,
next_cursor=None,
items=_audit_items(session, rows, access_admin),
items=_audit_items(session, rows, context.access_admin),
deleted=[],
watermark=_audit_delta_response_watermark(
session,
effective_scope=effective_scope,
tenant_id=resolved_tenant_id,
effective_scope=context.effective_scope,
tenant_id=context.resolved_tenant_id,
entries=entries,
has_more=has_more,
),
@@ -608,3 +649,78 @@ def list_audit_log(
query = query.filter(AuditLog.object_id == object_id)
items = query.order_by(AuditLog.created_at.desc()).offset(offset).limit(limit).all()
return AuditLogListResponse(items=[AuditLogItemResponse.model_validate(item) for item in items])
@router.get(
"/admin/audit/event-delivery/metrics",
response_model=EventDeliveryMetricsResponse,
)
def event_delivery_metrics(
session: Session = Depends(get_session),
_principal: ApiPrincipal = Depends(
require_any_scope("system:audit:read")
),
):
outbox = platform_event_outbox(get_registry())
if outbox is None:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="Durable platform event delivery is not configured",
)
return EventDeliveryMetricsResponse.model_validate(
outbox.delivery_metrics(session)
)
@router.post(
"/admin/audit/event-deliveries/{event_id}/{consumer_id}/replay",
response_model=EventDeliveryReplayResponse,
)
def replay_event_delivery(
event_id: str,
consumer_id: str,
payload: EventDeliveryReplayRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(
require_any_scope("system:governance:write")
),
):
outbox = platform_event_outbox(get_registry())
if outbox is None:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="Durable platform event delivery is not configured",
)
try:
result = outbox.replay_delivery(
session,
event_id=event_id,
consumer_id=consumer_id,
operator_id=principal.account_id,
reason=payload.reason,
)
except LookupError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=str(exc),
) from exc
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=str(exc),
) from exc
audit_from_principal(
session,
principal,
action="platform_event.delivery_replayed",
scope="system",
object_type="platform_event_delivery",
object_id=f"{event_id}:{consumer_id}",
details={
"event_id": event_id,
"consumer_id": consumer_id,
"reason": payload.reason,
},
)
session.commit()
return EventDeliveryReplayResponse.model_validate(result)
+27 -1
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
from datetime import datetime
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator
from pydantic import BaseModel, ConfigDict, Field
from govoplan_core.api.v1.schemas import DeltaDeletedItem
@@ -53,3 +53,29 @@ class AuditLogItemResponse(BaseModel):
class AuditLogListResponse(BaseModel):
items: list[AuditLogItemResponse]
class EventDeliveryMetricsResponse(BaseModel):
events: dict[str, int] = Field(default_factory=dict)
deliveries: dict[str, int] = Field(default_factory=dict)
consumers: dict[str, dict[str, int]] = Field(default_factory=dict)
oldest_due_at: datetime | None = None
class EventDeliveryReplayRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
reason: str = Field(min_length=1, max_length=2000)
class EventDeliveryReplayResponse(BaseModel):
event_id: str
consumer_id: str
delivery_key: str
status: str
attempts: int
replay_count: int
last_replayed_at: datetime | None = None
last_replayed_by: str | None = None
last_replay_reason: str | None = None
last_error: str | None = None
+53
View File
@@ -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)
+125 -2
View File
@@ -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,125 @@ 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)
class AuditOutboxDelivery(Base, TimestampMixin):
__tablename__ = "audit_outbox_deliveries"
__table_args__ = (
UniqueConstraint(
"outbox_event_id",
"consumer_id",
name="uq_audit_outbox_delivery_consumer",
),
UniqueConstraint(
"delivery_key",
name="uq_audit_outbox_delivery_key",
),
Index(
"ix_audit_outbox_delivery_status_next_attempt_at",
"status",
"next_attempt_at",
),
Index(
"ix_audit_outbox_delivery_consumer_status",
"consumer_id",
"status",
),
)
id: Mapped[str] = mapped_column(
String(36),
primary_key=True,
default=new_uuid,
)
outbox_event_id: Mapped[str] = mapped_column(
ForeignKey("audit_outbox_events.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
consumer_id: Mapped[str] = mapped_column(
String(128),
nullable=False,
)
delivery_key: Mapped[str] = mapped_column(
String(300),
nullable=False,
)
policy_decision_ref: Mapped[str | None] = mapped_column(
String(128),
nullable=True,
)
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,
)
delivered_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
quarantined_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
replay_count: Mapped[int] = mapped_column(
Integer,
nullable=False,
default=0,
)
last_replayed_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
last_replayed_by: Mapped[str | None] = mapped_column(
String(128),
nullable=True,
)
last_replay_reason: Mapped[str | None] = mapped_column(
Text,
nullable=True,
)
last_error: Mapped[str | None] = mapped_column(
Text,
nullable=True,
)
__all__ = [
"AuditLog",
"AuditOutboxDelivery",
"AuditOutboxEvent",
"new_uuid",
]
+92 -4
View File
@@ -1,5 +1,7 @@
from __future__ import annotations
from pathlib import Path
from govoplan_audit.backend.db import models as audit_models # noqa: F401 - populate Audit ORM metadata
from govoplan_core.core.access import (
CAPABILITY_AUDIT_RECORDER,
@@ -8,7 +10,10 @@ from govoplan_core.core.access import (
CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
)
from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard
from govoplan_core.core.modules import MigrationSpec, ModuleContext, ModuleManifest
from govoplan_core.core.modules import DocumentationTopic, FrontendModule, MigrationSpec, ModuleContext, ModuleManifest
from govoplan_core.core.provider_governance import declared_module_architecture
from govoplan_core.core.events import CAPABILITY_PLATFORM_EVENT_OUTBOX
from govoplan_core.core.views import ViewSurface
from govoplan_core.db.base import Base
@@ -33,26 +38,109 @@ def _audit_retention(context: ModuleContext):
return SqlAuditRetentionProvider()
def _event_outbox(context: ModuleContext):
from govoplan_audit.backend.outbox import SqlAuditOutbox
return SqlAuditOutbox(
max_attempts=getattr(
context.settings,
"platform_event_outbox_max_attempts",
8,
)
)
manifest = ModuleManifest(
id="audit",
name="Audit",
version="0.1.6",
version="0.1.17",
required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
route_factory=_route_factory,
documentation=(
DocumentationTopic(
id="audit.read-authorized-evidence",
title="Read authorized audit evidence",
summary="Audit history explains who performed a governed action, when it happened, and which resource and trace context were involved.",
body="Audit views are permission- and tenant-scoped. Entries are evidence, not editable business records. Sensitive payloads may be redacted while stable resource, actor, outcome, request, run, and trace references remain available for investigation.",
documentation_types=("user",),
audience=("auditor", "tenant_admin", "operator"),
metadata={
"kind": "reference",
"help_contexts": [
"audit.admin.system",
"audit.admin.tenant",
"audit.event-details",
],
"surfaces": ["audit.admin.system", "audit.admin.tenant"],
},
),
DocumentationTopic(
id="audit.recording-retention-and-outbox",
title="Operate audit recording and event delivery",
summary="Audit owns durable audit records, retention operations, and the transactional platform-event outbox.",
body="Modules record bounded audit facts through the Audit capability. Governed platform events are committed to the outbox with retry and delivery metadata so a failed consumer does not erase the originating transaction. Worker dispatch is partitioned by tenant entitlement; an unavailable consumer retains its durable delivery and records an operator-required outcome instead of acknowledging the event. Retention and destructive retirement must preserve the configured evidence and recovery guarantees.",
documentation_types=("admin",),
audience=("auditor", "security_officer", "operator"),
related_modules=("policy", "ops"),
metadata={
"kind": "reference",
"help_contexts": [
"audit.recording",
"audit.retention",
"audit.event-outbox",
],
},
),
),
frontend=FrontendModule(
module_id="audit",
package_name="@govoplan/audit-webui",
view_surfaces=(
ViewSurface(id="audit.admin.system", module_id="audit", kind="section", label="System audit", order=90),
ViewSurface(id="audit.admin.tenant", module_id="audit", kind="section", label="Tenant audit", order=100),
),
),
migration_spec=MigrationSpec(
module_id="audit",
metadata=Base.metadata,
script_location=str(
Path(__file__).with_name("migrations") / "versions"
),
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.AuditOutboxDelivery,
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.AuditOutboxDelivery,
audit_models.AuditOutboxEvent,
label="Audit",
),
),
capability_factories={
CAPABILITY_AUDIT_RECORDER: _audit_recorder,
CAPABILITY_AUDIT_RETENTION: _audit_retention,
CAPABILITY_PLATFORM_EVENT_OUTBOX: _event_outbox,
},
architecture=declared_module_architecture(
layer="governance_accountability",
kind="governance",
maturity="vertical_slice",
documentation_ref="docs/AUDIT_TRACE_CONTEXT.md",
test_ref="tests/test_audit_module_contract.py",
known_limits=("Cross-deployment archival and evidentiary export profiles are not yet reference-ready.",),
owned_concepts=("audit record", "audit retention", "transactional event outbox"),
non_owned_concepts=("domain record", "policy decision", "external effect"),
recovery_docs=("README.md",),
security_docs=("docs/AUDIT_TRACE_CONTEXT.md",),
operations_docs=("README.md",),
),
)
@@ -0,0 +1 @@
"""Audit module database migrations."""
@@ -0,0 +1 @@
"""Development-track Audit migrations."""
@@ -0,0 +1,92 @@
"""durable platform event delivery ledger
Revision ID: a8d1e4f7b2c5
Revises: None
Create Date: 2026-07-29 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "a8d1e4f7b2c5"
down_revision = None
branch_labels = None
depends_on = "c91f0a72be34"
def upgrade() -> None:
inspector = sa.inspect(op.get_bind())
if "audit_outbox_deliveries" in inspector.get_table_names():
return
op.create_table(
"audit_outbox_deliveries",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("outbox_event_id", sa.String(length=36), nullable=False),
sa.Column("consumer_id", sa.String(length=128), nullable=False),
sa.Column("delivery_key", sa.String(length=300), nullable=False),
sa.Column("policy_decision_ref", sa.String(length=128), nullable=True),
sa.Column("status", sa.String(length=20), nullable=False),
sa.Column("attempts", sa.Integer(), nullable=False),
sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("delivered_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("quarantined_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("replay_count", sa.Integer(), nullable=False),
sa.Column("last_replayed_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("last_replayed_by", sa.String(length=128), nullable=True),
sa.Column("last_replay_reason", sa.Text(), nullable=True),
sa.Column("last_error", sa.Text(), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.ForeignKeyConstraint(
["outbox_event_id"],
["audit_outbox_events.id"],
name=op.f(
"fk_audit_outbox_deliveries_outbox_event_id_"
"audit_outbox_events"
),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_audit_outbox_deliveries"),
),
sa.UniqueConstraint(
"delivery_key",
name="uq_audit_outbox_delivery_key",
),
sa.UniqueConstraint(
"outbox_event_id",
"consumer_id",
name="uq_audit_outbox_delivery_consumer",
),
)
op.create_index(
"ix_audit_outbox_deliveries_outbox_event_id",
"audit_outbox_deliveries",
["outbox_event_id"],
)
op.create_index(
"ix_audit_outbox_deliveries_status",
"audit_outbox_deliveries",
["status"],
)
op.create_index(
"ix_audit_outbox_delivery_status_next_attempt_at",
"audit_outbox_deliveries",
["status", "next_attempt_at"],
)
op.create_index(
"ix_audit_outbox_delivery_consumer_status",
"audit_outbox_deliveries",
["consumer_id", "status"],
)
def downgrade() -> None:
if (
"audit_outbox_deliveries"
in sa.inspect(op.get_bind()).get_table_names()
):
op.drop_table("audit_outbox_deliveries")
@@ -0,0 +1 @@
"""Release-track Audit migrations."""
@@ -0,0 +1,92 @@
"""durable platform event delivery ledger
Revision ID: a8d1e4f7b2c5
Revises: None
Create Date: 2026-07-29 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "a8d1e4f7b2c5"
down_revision = None
branch_labels = None
depends_on = "c91f0a72be34"
def upgrade() -> None:
inspector = sa.inspect(op.get_bind())
if "audit_outbox_deliveries" in inspector.get_table_names():
return
op.create_table(
"audit_outbox_deliveries",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("outbox_event_id", sa.String(length=36), nullable=False),
sa.Column("consumer_id", sa.String(length=128), nullable=False),
sa.Column("delivery_key", sa.String(length=300), nullable=False),
sa.Column("policy_decision_ref", sa.String(length=128), nullable=True),
sa.Column("status", sa.String(length=20), nullable=False),
sa.Column("attempts", sa.Integer(), nullable=False),
sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("delivered_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("quarantined_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("replay_count", sa.Integer(), nullable=False),
sa.Column("last_replayed_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("last_replayed_by", sa.String(length=128), nullable=True),
sa.Column("last_replay_reason", sa.Text(), nullable=True),
sa.Column("last_error", sa.Text(), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.ForeignKeyConstraint(
["outbox_event_id"],
["audit_outbox_events.id"],
name=op.f(
"fk_audit_outbox_deliveries_outbox_event_id_"
"audit_outbox_events"
),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_audit_outbox_deliveries"),
),
sa.UniqueConstraint(
"delivery_key",
name="uq_audit_outbox_delivery_key",
),
sa.UniqueConstraint(
"outbox_event_id",
"consumer_id",
name="uq_audit_outbox_delivery_consumer",
),
)
op.create_index(
"ix_audit_outbox_deliveries_outbox_event_id",
"audit_outbox_deliveries",
["outbox_event_id"],
)
op.create_index(
"ix_audit_outbox_deliveries_status",
"audit_outbox_deliveries",
["status"],
)
op.create_index(
"ix_audit_outbox_delivery_status_next_attempt_at",
"audit_outbox_deliveries",
["status", "next_attempt_at"],
)
op.create_index(
"ix_audit_outbox_delivery_consumer_status",
"audit_outbox_deliveries",
["consumer_id", "status"],
)
def downgrade() -> None:
if (
"audit_outbox_deliveries"
in sa.inspect(op.get_bind()).get_table_names()
):
op.drop_table("audit_outbox_deliveries")
+659
View File
@@ -0,0 +1,659 @@
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")))
+420
View File
@@ -0,0 +1,420 @@
from __future__ import annotations
import unittest
from datetime import datetime, timedelta, timezone
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 (
AuditOutboxDelivery,
AuditOutboxEvent,
)
from govoplan_audit.backend.outbox import SqlAuditOutbox
from govoplan_core.core.events import (
DurableEventConsumer,
EventActorRef,
EventTenantRef,
PlatformEvent,
)
from govoplan_core.core.institutional import (
GovernedContextEnvelope,
InstitutionalReference,
TemporalRevision,
)
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 _database(self):
engine = create_engine("sqlite:///:memory:")
self.addCleanup(engine.dispose)
Base.metadata.create_all(
bind=engine,
tables=[
AuditOutboxEvent.__table__,
AuditOutboxDelivery.__table__,
],
)
return sessionmaker(bind=engine)
def test_outbox_enqueues_governed_event_and_dispatches_pending_rows(self) -> None:
Session = self._database()
outbox = SqlAuditOutbox()
seen: list[PlatformEvent] = []
observed: 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,
consumers=(
DurableEventConsumer(
consumer_id="tests.consumer.v1",
event_types=frozenset({"tenant.created"}),
handler=lambda delivered, _key: seen.append(
delivered
),
),
),
observer=observed.append,
)
self.assertEqual(
{
"selected": 1,
"delivered": 1,
"retrying": 0,
"quarantined": 0,
"dispatched": 1,
"observer_failed": 0,
},
counts,
)
self.assertEqual(1, len(seen))
self.assertEqual(1, len(observed))
self.assertEqual("tenant.created", seen[0].type)
self.assertEqual("dispatched", row.status)
self.assertEqual(1, row.attempts)
self.assertIsNotNone(row.dispatched_at)
delivery = session.query(AuditOutboxDelivery).one()
self.assertEqual("delivered", delivery.status)
self.assertEqual(
f"{event.event_id}:tests.consumer.v1",
delivery.delivery_key,
)
def test_dispatch_partitions_pending_events_by_tenant(self) -> None:
Session = self._database()
outbox = SqlAuditOutbox()
seen: list[str] = []
with Session() as session:
first = outbox.enqueue(
session,
PlatformEvent(
type="files.file.created",
module_id="files",
tenant=EventTenantRef(id="tenant-1"),
),
)
second = outbox.enqueue(
session,
PlatformEvent(
type="files.file.created",
module_id="files",
tenant=EventTenantRef(id="tenant-2"),
),
)
counts = outbox.dispatch_pending(
session,
tenant_id="tenant-1",
consumers=(
DurableEventConsumer(
consumer_id="tests.tenant-filter.v1",
handler=lambda event, _key: seen.append(
event.tenant.id if event.tenant else "system"
),
),
),
observer=None,
)
self.assertEqual(1, counts["selected"])
self.assertEqual(["tenant-1"], seen)
self.assertEqual("dispatched", first.status)
self.assertEqual("pending", second.status)
def test_dispatch_can_select_only_tenantless_system_events(self) -> None:
Session = self._database()
outbox = SqlAuditOutbox()
seen: list[str] = []
with Session() as session:
system = outbox.enqueue(
session,
PlatformEvent(type="system.ready", module_id="core"),
)
tenant = outbox.enqueue(
session,
PlatformEvent(
type="tenant.ready",
module_id="tenancy",
tenant=EventTenantRef(id="tenant-1"),
),
)
counts = outbox.dispatch_pending(
session,
tenantless_only=True,
consumers=(
DurableEventConsumer(
consumer_id="tests.system-filter.v1",
handler=lambda event, _key: seen.append(event.type),
),
),
observer=None,
)
self.assertEqual(1, counts["selected"])
self.assertEqual(["system.ready"], seen)
self.assertEqual("dispatched", system.status)
self.assertEqual("pending", tenant.status)
def test_outbox_preserves_institutional_context(self) -> None:
Session = self._database()
outbox = SqlAuditOutbox()
seen: list[PlatformEvent] = []
now = datetime.now(timezone.utc)
context = GovernedContextEnvelope(
tenant_id="tenant-1",
temporal=TemporalRevision(revision="decision:7", recorded_at=now),
decision_ref=InstitutionalReference(
kind="decision",
owner_module="committee",
object_id="decision-7",
tenant_id="tenant-1",
version="7",
valid_at=now,
),
approval_refs=(
InstitutionalReference(
kind="approval",
owner_module="workflow",
object_id="approval-3",
tenant_id="tenant-1",
version="3",
valid_at=now,
),
),
)
with Session() as session:
outbox.enqueue(
session,
PlatformEvent(
type="committee.decision.recorded",
module_id="committee",
institutional_context=context,
),
)
outbox.dispatch_pending(
session,
consumers=(
DurableEventConsumer(
consumer_id="tests.institutional-context.v1",
event_types=frozenset({"committee.decision.recorded"}),
handler=lambda delivered, _key: seen.append(delivered),
),
),
)
self.assertEqual("decision-7", seen[0].institutional_context.decision_ref.object_id)
self.assertEqual(
"approval-3",
seen[0].institutional_context.approval_refs[0].object_id,
)
def test_outbox_retries_then_quarantines_a_failed_consumer(self) -> None:
Session = self._database()
outbox = SqlAuditOutbox(max_attempts=2)
consumer = DurableEventConsumer(
consumer_id="tests.failing.v1",
handler=lambda _event, _key: (_ for _ in ()).throw(
RuntimeError("offline")
),
)
with Session() as session:
row = outbox.enqueue(session, PlatformEvent(type="demo.failed", module_id="audit"))
counts = outbox.dispatch_pending(
session,
consumers=(consumer,),
observer=None,
)
self.assertEqual(1, counts["retrying"])
self.assertEqual("retrying", row.status)
self.assertEqual(1, row.attempts)
self.assertEqual("offline", row.last_error)
self.assertIsNotNone(row.next_attempt_at)
delivery = session.query(AuditOutboxDelivery).one()
delivery.next_attempt_at = None
row.next_attempt_at = None
second = outbox.dispatch_pending(
session,
consumers=(consumer,),
observer=None,
)
self.assertEqual(1, second["quarantined"])
self.assertEqual("quarantined", row.status)
self.assertEqual("quarantined", delivery.status)
self.assertIsNotNone(delivery.quarantined_at)
self.assertIsNone(delivery.next_attempt_at)
def test_replay_keeps_a_stable_delivery_key_and_runs_once(self) -> None:
Session = self._database()
outbox = SqlAuditOutbox(max_attempts=1)
event = PlatformEvent(type="demo.replay", module_id="audit")
failing = DurableEventConsumer(
consumer_id="tests.replay.v1",
handler=lambda _event, _key: (_ for _ in ()).throw(
RuntimeError("offline")
),
)
delivered: list[str] = []
with Session() as session:
outbox.enqueue(session, event)
outbox.dispatch_pending(
session,
consumers=(failing,),
observer=None,
)
state = outbox.replay_delivery(
session,
event_id=event.event_id,
consumer_id=failing.consumer_id,
operator_id="operator-1",
reason="Dependency recovered",
)
self.assertEqual("pending", state["status"])
self.assertEqual(1, state["replay_count"])
expected_key = f"{event.event_id}:{failing.consumer_id}"
self.assertEqual(expected_key, state["delivery_key"])
healthy = DurableEventConsumer(
consumer_id=failing.consumer_id,
handler=lambda _event, key: delivered.append(key),
)
outbox.dispatch_pending(
session,
consumers=(healthy,),
observer=None,
)
replay = outbox.dispatch_pending(
session,
consumers=(healthy,),
observer=None,
)
self.assertEqual([expected_key], delivered)
self.assertEqual(0, replay["selected"])
def test_classified_subscription_requires_and_persists_policy_decision(self) -> None:
with self.assertRaisesRegex(ValueError, "policy decision"):
DurableEventConsumer(
consumer_id="tests.restricted.v1",
classifications=frozenset({"restricted"}),
handler=lambda _event, _key: None,
)
Session = self._database()
outbox = SqlAuditOutbox()
consumer = DurableEventConsumer(
consumer_id="tests.restricted.v1",
classifications=frozenset({"restricted"}),
policy_decision_ref="policy-decision:42",
handler=lambda _event, _key: None,
)
with Session() as session:
outbox.enqueue(
session,
PlatformEvent(
type="case.changed",
module_id="cases",
classification="restricted",
),
)
outbox.dispatch_pending(
session,
consumers=(consumer,),
observer=None,
)
delivery = session.query(AuditOutboxDelivery).one()
self.assertEqual(
"policy-decision:42",
delivery.policy_decision_ref,
)
def test_metrics_and_retention_keep_quarantined_evidence(self) -> None:
Session = self._database()
outbox = SqlAuditOutbox(max_attempts=1)
with Session() as session:
delivered_event = PlatformEvent(
type="demo.delivered",
module_id="audit",
)
failed_event = PlatformEvent(
type="demo.failed",
module_id="audit",
)
delivered_row = outbox.enqueue(session, delivered_event)
outbox.enqueue(session, failed_event)
consumer = DurableEventConsumer(
consumer_id="tests.metrics.v1",
handler=lambda event, _key: (
(_ for _ in ()).throw(RuntimeError("offline"))
if event.type == "demo.failed"
else None
),
)
outbox.dispatch_pending(
session,
consumers=(consumer,),
observer=None,
)
delivered_row.dispatched_at = (
datetime.now(timezone.utc) - timedelta(days=100)
)
metrics = outbox.delivery_metrics(session)
purged = outbox.purge_terminal(
session,
before=datetime.now(timezone.utc)
- timedelta(days=90),
)
self.assertEqual(1, metrics["events"]["dispatched"])
self.assertEqual(1, metrics["events"]["quarantined"])
self.assertEqual(1, purged["deleted"])
remaining = session.query(AuditOutboxEvent).one()
self.assertEqual("quarantined", remaining.status)
if __name__ == "__main__":
unittest.main()
+45
View File
@@ -0,0 +1,45 @@
from __future__ import annotations
import pathlib
import tomllib
import unittest
from govoplan_audit.backend.manifest import get_manifest
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.8", 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)
def test_audit_documentation_declares_admin_help_contexts(self) -> None:
topics = {topic.id: topic for topic in get_manifest().documentation}
evidence = topics["audit.read-authorized-evidence"]
self.assertIn("audit.admin.system", evidence.metadata["help_contexts"])
self.assertIn("audit.admin.tenant", evidence.metadata["help_contexts"])
self.assertEqual(
["audit.admin.system", "audit.admin.tenant"],
evidence.metadata["surfaces"],
)
operations = topics["audit.recording-retention-and-outbox"]
self.assertIn("audit.retention", operations.metadata["help_contexts"])
if __name__ == "__main__":
unittest.main()
+30
View File
@@ -0,0 +1,30 @@
{
"name": "@govoplan/audit-webui",
"version": "0.1.17",
"private": true,
"type": "module",
"main": "src/index.ts",
"module": "src/index.ts",
"types": "src/index.ts",
"exports": {
".": {
"types": "./src/index.ts",
"import": "./src/index.ts"
}
},
"scripts": {
"test:interface-patterns": "node scripts/test-interface-pattern-language.mjs"
},
"peerDependencies": {
"@govoplan/core-webui": "^0.1.17",
"lucide-react": "^1.23.0",
"react": ">=19.2.7 <20",
"react-dom": ">=19.2.7 <20",
"react-router": ">=8.3.0 <9"
},
"peerDependenciesMeta": {
"@govoplan/core-webui": {
"optional": true
}
}
}
@@ -0,0 +1,29 @@
import assert from "node:assert/strict";
import { readFileSync } from "node:fs";
import { fileURLToPath } from "node:url";
const panel = readFileSync(
fileURLToPath(new URL("../src/features/audit/AdminAuditPanel.tsx", import.meta.url)),
"utf8"
);
const moduleSource = readFileSync(
fileURLToPath(new URL("../src/module.ts", import.meta.url)),
"utf8"
);
assert.match(panel, /AdminPageLayout,[\s\S]*DataGrid,[\s\S]*Dialog,[\s\S]*DocumentationHelpLink,[\s\S]*TableActionGroup/);
assert.match(panel, /topicId: "audit\.read-authorized-evidence"/);
assert.match(panel, /disabledReason=\{loading \? I18N\.loading : undefined\}/);
assert.match(panel, /pagination=\{\{[\s\S]*mode: "server"/);
assert.match(panel, /id="admin-audit-event-details-grid"/);
assert.match(panel, /auditDetailRows\(selected\?\.details \?\? \{\}\)/);
assert.match(panel, /emptyText=\{I18N\.noRecords\}/);
assert.doesNotMatch(panel, /<pre|window\.(?:alert|confirm)\(/);
assert.doesNotMatch(panel, /@govoplan\/(?:access|admin)-webui|govoplan_(?:access|admin)/);
assert.match(moduleSource, /generatedTranslations/);
assert.match(moduleSource, /translations,/);
assert.match(moduleSource, /surfaceId: "audit\.admin\.system"[\s\S]*allOf: \["system:audit:read"\]/);
assert.match(moduleSource, /surfaceId: "audit\.admin\.tenant"[\s\S]*allOf: \["audit:read"\]/);
console.log("Audit administration surfaces satisfy the monitoring and evidence pattern contracts.");
+74
View File
@@ -0,0 +1,74 @@
import { apiFetch, type ApiSettings, type DeltaDeletedItem } from "@govoplan/core-webui";
export type AuditAdminItem = {
id: string;
scope: "tenant" | "system";
tenant_id?: string | null;
actor_email?: string | null;
action: string;
object_type?: string | null;
object_id?: string | null;
details: Record<string, unknown>;
created_at: string;
};
export type AuditSortBy = "time" | "actor" | "action" | "object" | "tenant";
export type AuditQueryOptions = {
tenantId?: string | null;
allTenants?: boolean;
scope?: "tenant" | "system";
limit?: number;
offset?: number;
page?: number;
pageSize?: number;
cursor?: string | null;
sortBy?: AuditSortBy;
sortDirection?: "asc" | "desc";
filters?: Partial<Record<AuditSortBy, string>>;
};
export type AuditAdminListResponse = {
items: AuditAdminItem[];
total: number;
page: number;
page_size: number;
pages: number;
cursor?: string | null;
next_cursor?: string | null;
};
export type AuditAdminDeltaResponse = AuditAdminListResponse & {
deleted: DeltaDeletedItem[];
watermark?: string | null;
has_more: boolean;
full: boolean;
};
function auditQuery(options: AuditQueryOptions & { since?: string | null } = {}): string {
const params = new URLSearchParams();
if (options.tenantId) params.set("tenant_id", options.tenantId);
if (options.allTenants) params.set("all_tenants", "true");
if (options.scope) params.set("scope", options.scope);
if (options.limit) params.set("limit", String(options.limit));
if (options.offset) params.set("offset", String(options.offset));
if (options.page) params.set("page", String(options.page));
if (options.pageSize) params.set("page_size", String(options.pageSize));
if (options.cursor) params.set("cursor", options.cursor);
if (options.sortBy) params.set("sort_by", options.sortBy);
if (options.sortDirection) params.set("sort_direction", options.sortDirection);
if (options.since) params.set("since", options.since);
for (const [column, value] of Object.entries(options.filters ?? {})) {
if (value?.trim()) params.set(`filter_${column}`, value);
}
const suffix = params.toString();
return suffix ? `?${suffix}` : "";
}
export function fetchAdminAudit(settings: ApiSettings, options: AuditQueryOptions = {}): Promise<AuditAdminListResponse> {
return apiFetch(settings, `/api/v1/admin/audit${auditQuery(options)}`);
}
export function fetchAdminAuditDelta(settings: ApiSettings, options: AuditQueryOptions & { since?: string | null } = {}): Promise<AuditAdminDeltaResponse> {
return apiFetch(settings, `/api/v1/admin/audit/delta${auditQuery(options)}`);
}
@@ -0,0 +1,281 @@
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import { Search } from "lucide-react";
import {
AdminPageLayout,
adminErrorMessage,
Button,
DataGrid,
Dialog,
DocumentationHelpLink,
formatAdminDateTime as formatDateTime,
i18nMessage,
mergeDeltaRows,
TableActionGroup,
useDeltaWatermarks,
type ApiSettings,
type AuthInfo,
type DataGridColumn,
type DataGridQueryState
} from "@govoplan/core-webui";
import { fetchAdminAudit, fetchAdminAuditDelta, type AuditAdminItem, type AuditSortBy } from "../../api/audit";
type Props = {
settings: ApiSettings;
auth: AuthInfo;
systemMode?: boolean;
};
type AuditDetailRow = {
id: string;
field: string;
value: string;
};
const I18N = {
actionLabel: "i18n:govoplan-audit.action.f1a20801",
actions: "i18n:govoplan-audit.actions.f1a20802",
actor: "i18n:govoplan-audit.actor.f1a20803",
close: "i18n:govoplan-audit.close.f1a20804",
details: "i18n:govoplan-audit.details.f1a20805",
eventDetails: "i18n:govoplan-audit.audit_event_details.f1a20806",
inspect: "i18n:govoplan-audit.inspect_audit_event.f1a20807",
loading: "i18n:govoplan-audit.audit_evidence_is_loading.f1a20808",
noDetails: "i18n:govoplan-audit.no_additional_details_were_recorded.f1a20809",
noRecords: "i18n:govoplan-audit.no_audit_records_match_the_current_scope_and_filters.f1a20810",
object: "i18n:govoplan-audit.object.f1a20811",
reload: "i18n:govoplan-audit.reload_audit_evidence.f1a20812",
scopeLabel: "i18n:govoplan-audit.scope.f1a20813",
system: "i18n:govoplan-audit.system.f1a20814",
systemAudit: "i18n:govoplan-audit.system_audit.f1a20815",
systemDescription: "i18n:govoplan-audit.system_level_administrative_history_showing_value0_value1_of_value2.f1a20816",
tenantAudit: "i18n:govoplan-audit.tenant_audit.f1a20817",
tenantContext: "i18n:govoplan-audit.tenant_context.f1a20818",
tenantDescription: "i18n:govoplan-audit.tenant_level_administrative_history_showing_value0_value1_of_value2.f1a20819",
time: "i18n:govoplan-audit.time.f1a20820",
value: "i18n:govoplan-audit.value.f1a20821"
} as const;
const DEFAULT_QUERY: DataGridQueryState = {
sort: { columnId: "time", direction: "desc" },
filters: {}
};
export default function AdminAuditPanel({ settings, auth, systemMode = false }: Props) {
const [items, setItems] = useState<AuditAdminItem[]>([]);
const itemsRef = useRef<AuditAdminItem[]>([]);
const pageItemsRef = useRef<Record<string, AuditAdminItem[]>>({});
const pageCursorsRef = useRef<Record<number, string | null>>({ 1: null });
const { getDeltaWatermark, setDeltaWatermark, resetDeltaWatermark } = useDeltaWatermarks();
const [total, setTotal] = useState(0);
const [page, setPage] = useState(1);
const [pageSize, setPageSize] = useState(10);
const [query, setQuery] = useState<DataGridQueryState>(DEFAULT_QUERY);
const [selected, setSelected] = useState<AuditAdminItem | null>(null);
const [loading, setLoading] = useState(true);
const [error, setError] = useState("");
const [reloadToken, setReloadToken] = useState(0);
const tenantId = (auth.active_tenant ?? auth.tenant).id;
const load = useCallback(async () => {
setLoading(true);
setError("");
try {
const sortColumn = query.sort?.columnId;
const sortBy = sortColumn && ["time", "actor", "action", "object", "tenant"].includes(sortColumn)
? sortColumn as AuditSortBy
: "time";
const sortDirection = query.sort?.direction ?? "desc";
const filters = query.filters;
const pageCursor = page === 1 ? null : pageCursorsRef.current[page];
const deltaMode = page === 1 || pageCursor !== undefined;
const requestOptions = {
scope: systemMode ? "system" as const : "tenant" as const,
page,
pageSize,
cursor: pageCursor,
sortBy,
sortDirection,
filters
};
const deltaKey = `audit:${systemMode ? "system" : "tenant"}:${tenantId}:${pageSize}:${page}:${pageCursor ?? "root"}:${JSON.stringify({ sortBy, sortDirection, filters })}`;
const response = deltaMode
? await fetchAdminAuditDelta(settings, { ...requestOptions, since: getDeltaWatermark(deltaKey) })
: await fetchAdminAudit(settings, requestOptions);
const baseItems = pageItemsRef.current[deltaKey] ?? [];
const nextItems = "full" in response && !response.full
? mergeDeltaRows(baseItems, response.items, response.deleted, (item) => item.id, { sort: compareAuditEvents(sortBy, sortDirection) }).slice(0, pageSize)
: response.items;
pageItemsRef.current[deltaKey] = nextItems;
itemsRef.current = nextItems;
setItems(nextItems);
setTotal(response.total);
if (!deltaMode && response.page !== page) setPage(response.page);
if (response.cursor !== undefined) pageCursorsRef.current[page] = response.cursor ?? null;
if (response.next_cursor !== undefined) {
if (response.next_cursor) pageCursorsRef.current[page + 1] = response.next_cursor;
else delete pageCursorsRef.current[page + 1];
}
if ("full" in response && !response.full && page === 1 && (response.items.length > 0 || response.deleted.length > 0)) {
pageCursorsRef.current = { 1: null };
}
if ("watermark" in response) setDeltaWatermark(deltaKey, response.watermark);
if (!deltaMode) resetDeltaWatermark(deltaKey);
} catch (err) {
setError(adminErrorMessage(err));
} finally {
setLoading(false);
}
}, [settings.accessToken, settings.apiBaseUrl, settings.apiKey, systemMode, tenantId, page, pageSize, query, reloadToken, getDeltaWatermark, setDeltaWatermark, resetDeltaWatermark]);
useEffect(() => {
itemsRef.current = [];
pageItemsRef.current = {};
pageCursorsRef.current = { 1: null };
resetDeltaWatermark();
}, [settings.accessToken, settings.apiBaseUrl, settings.apiKey, systemMode, tenantId, pageSize, query, resetDeltaWatermark]);
useEffect(() => { void load(); }, [load]);
const handleQueryChange = useCallback((next: DataGridQueryState) => {
setQuery((current) => {
if (JSON.stringify(current) === JSON.stringify(next)) return current;
setPage(1);
return next;
});
}, []);
const columns = useMemo<DataGridColumn<AuditAdminItem>[]>(() => [
{ id: "time", header: I18N.time, width: 190, minWidth: 150, maxWidth: 260, resizable: true, sticky: "start", sortable: true, filterable: true, filterType: "date", value: (row) => row.created_at, render: (row) => formatDateTime(row.created_at) },
{ id: "actor", header: I18N.actor, width: 220, minWidth: 170, maxWidth: 360, resizable: true, sortable: true, filterable: true, value: (row) => row.actor_email || "System", render: (row) => row.actor_email || I18N.system },
{ id: "action", header: I18N.actionLabel, width: 250, minWidth: 170, maxWidth: 420, resizable: true, sortable: true, filterable: true, value: (row) => row.action },
{ id: "object", header: I18N.object, width: 300, minWidth: 180, maxWidth: 640, resizable: true, fill: true, sortable: true, filterable: true, value: (row) => `${row.object_type || "-"} ${row.object_id || ""}`.trim() },
...(systemMode ? [{ id: "tenant", header: I18N.tenantContext, width: 190, minWidth: 150, maxWidth: 300, resizable: true, sortable: true, filterable: true, value: (row: AuditAdminItem) => row.tenant_id || "-" }] : []),
{ id: "actions", header: I18N.actions, width: 70, sticky: "end", resizable: false, align: "right", render: (row) => <TableActionGroup actions={[{ id: "inspect", label: I18N.inspect, icon: <Search aria-hidden="true" size={16} />, onClick: () => setSelected(row) }]} /> }
], [systemMode]);
const detailColumns = useMemo<DataGridColumn<AuditDetailRow>[]>(() => [
{ id: "field", header: I18N.details, minWidth: 180, resizable: true, value: (row) => row.field },
{ id: "value", header: I18N.value, minWidth: 260, resizable: true, fill: true, value: (row) => row.value }
], []);
const detailRows = useMemo(() => auditDetailRows(selected?.details ?? {}), [selected]);
const firstShown = total === 0 ? 0 : (page - 1) * pageSize + 1;
const lastShown = Math.min(total, page * pageSize);
const pageDescription = i18nMessage(
systemMode ? I18N.systemDescription : I18N.tenantDescription,
{ value0: firstShown, value1: lastShown, value2: total }
);
return (
<>
<AdminPageLayout
title={systemMode ? I18N.systemAudit : I18N.tenantAudit}
description={pageDescription}
loading={loading}
error={error}
actions={(
<>
<DocumentationHelpLink
reference={{
topicId: "audit.read-authorized-evidence",
documentationType: "user"
}} />
<Button
onClick={() => setReloadToken((value) => value + 1)}
disabled={loading}
disabledReason={loading ? I18N.loading : undefined}>
{I18N.reload}
</Button>
</>
)}>
<div className="admin-table-surface">
<DataGrid
id={systemMode ? "admin-system-audit-v6" : "admin-tenant-audit-v6"}
rows={items}
columns={columns}
initialFit="container"
getRowKey={(row) => row.id}
emptyText={I18N.noRecords}
className="admin-audit-grid"
initialSort={{ columnId: "time", direction: "desc" }}
pagination={{
mode: "server",
page,
pageSize,
totalRows: total,
pageSizeOptions: [10, 25, 50, 100, 250],
disabled: loading,
onPageChange: setPage,
onPageSizeChange: (next) => { setPageSize(next); setPage(1); }
}}
onQueryChange={handleQueryChange}
/>
</div>
</AdminPageLayout>
<Dialog
open={Boolean(selected)}
title={I18N.eventDetails}
onClose={() => setSelected(null)}
className="admin-dialog admin-dialog-wide"
footer={<Button onClick={() => setSelected(null)}>{I18N.close}</Button>}>
{selected && (
<>
<dl className="admin-details-grid">
<div><dt>{I18N.scopeLabel}</dt><dd>{selected.scope}</dd></div>
<div><dt>{I18N.actionLabel}</dt><dd>{selected.action}</dd></div>
<div><dt>{I18N.actor}</dt><dd>{selected.actor_email || I18N.system}</dd></div>
<div><dt>{I18N.object}</dt><dd>{selected.object_type || "-"} {selected.object_id || ""}</dd></div>
<div><dt>{I18N.tenantContext}</dt><dd>{selected.tenant_id || "-"}</dd></div>
<div><dt>{I18N.time}</dt><dd>{formatDateTime(selected.created_at)}</dd></div>
</dl>
{detailRows.length ? (
<DataGrid
id="admin-audit-event-details-grid"
rows={detailRows}
columns={detailColumns}
getRowKey={(row) => row.id}
initialFit="container" />
) : <p className="muted">{I18N.noDetails}</p>}
</>
)}
</Dialog>
</>
);
}
function auditDetailRows(details: Record<string, unknown>): AuditDetailRow[] {
return Object.entries(details)
.sort(([left], [right]) => left.localeCompare(right))
.map(([field, value]) => ({
id: field,
field,
value: auditDetailValue(value)
}));
}
function auditDetailValue(value: unknown): string {
if (value === null || value === undefined) return "-";
if (typeof value === "string") return value;
if (typeof value === "number" || typeof value === "boolean") return String(value);
return JSON.stringify(value);
}
function compareAuditEvents(sortBy: AuditSortBy, sortDirection: "asc" | "desc"): (left: AuditAdminItem, right: AuditAdminItem) => number {
return (left, right) => {
const primary = compareAuditValues(auditSortValue(left, sortBy), auditSortValue(right, sortBy));
const directed = sortDirection === "asc" ? primary : -primary;
return directed || right.id.localeCompare(left.id);
};
}
function auditSortValue(item: AuditAdminItem, sortBy: AuditSortBy): string | number {
if (sortBy === "time") return new Date(item.created_at).getTime();
if (sortBy === "actor") return item.actor_email || "System";
if (sortBy === "action") return item.action;
if (sortBy === "object") return `${item.object_type || ""} ${item.object_id || ""}`;
return item.tenant_id || "";
}
function compareAuditValues(left: string | number, right: string | number): number {
if (typeof left === "number" && typeof right === "number") return left - right;
return String(left).localeCompare(String(right));
}
+50
View File
@@ -0,0 +1,50 @@
import type { PlatformTranslations } from "@govoplan/core-webui";
export const generatedTranslations: PlatformTranslations = {
en: {
"i18n:govoplan-audit.action.f1a20801": "Action",
"i18n:govoplan-audit.actions.f1a20802": "Actions",
"i18n:govoplan-audit.actor.f1a20803": "Actor",
"i18n:govoplan-audit.close.f1a20804": "Close",
"i18n:govoplan-audit.details.f1a20805": "Detail",
"i18n:govoplan-audit.audit_event_details.f1a20806": "Audit event details",
"i18n:govoplan-audit.inspect_audit_event.f1a20807": "Inspect audit event",
"i18n:govoplan-audit.audit_evidence_is_loading.f1a20808": "Audit evidence is loading.",
"i18n:govoplan-audit.no_additional_details_were_recorded.f1a20809": "No additional details were recorded.",
"i18n:govoplan-audit.no_audit_records_match_the_current_scope_and_filters.f1a20810": "No audit records match the current scope and filters.",
"i18n:govoplan-audit.object.f1a20811": "Object",
"i18n:govoplan-audit.reload_audit_evidence.f1a20812": "Reload audit evidence",
"i18n:govoplan-audit.scope.f1a20813": "Scope",
"i18n:govoplan-audit.system.f1a20814": "System",
"i18n:govoplan-audit.system_audit.f1a20815": "System audit",
"i18n:govoplan-audit.system_level_administrative_history_showing_value0_value1_of_value2.f1a20816": "System-level administrative history, showing {value0}-{value1} of {value2}.",
"i18n:govoplan-audit.tenant_audit.f1a20817": "Tenant audit",
"i18n:govoplan-audit.tenant_context.f1a20818": "Tenant context",
"i18n:govoplan-audit.tenant_level_administrative_history_showing_value0_value1_of_value2.f1a20819": "Tenant-level administrative history for the active tenant, showing {value0}-{value1} of {value2}.",
"i18n:govoplan-audit.time.f1a20820": "Time",
"i18n:govoplan-audit.value.f1a20821": "Value"
},
de: {
"i18n:govoplan-audit.action.f1a20801": "Aktion",
"i18n:govoplan-audit.actions.f1a20802": "Aktionen",
"i18n:govoplan-audit.actor.f1a20803": "Akteur",
"i18n:govoplan-audit.close.f1a20804": "Schließen",
"i18n:govoplan-audit.details.f1a20805": "Detail",
"i18n:govoplan-audit.audit_event_details.f1a20806": "Details des Auditereignisses",
"i18n:govoplan-audit.inspect_audit_event.f1a20807": "Auditereignis prüfen",
"i18n:govoplan-audit.audit_evidence_is_loading.f1a20808": "Auditnachweise werden geladen.",
"i18n:govoplan-audit.no_additional_details_were_recorded.f1a20809": "Es wurden keine zusätzlichen Details aufgezeichnet.",
"i18n:govoplan-audit.no_audit_records_match_the_current_scope_and_filters.f1a20810": "Keine Auditaufzeichnungen entsprechen dem aktuellen Bereich und den Filtern.",
"i18n:govoplan-audit.object.f1a20811": "Objekt",
"i18n:govoplan-audit.reload_audit_evidence.f1a20812": "Auditnachweise neu laden",
"i18n:govoplan-audit.scope.f1a20813": "Geltungsbereich",
"i18n:govoplan-audit.system.f1a20814": "System",
"i18n:govoplan-audit.system_audit.f1a20815": "Systemaudit",
"i18n:govoplan-audit.system_level_administrative_history_showing_value0_value1_of_value2.f1a20816": "Systemweite administrative Historie, angezeigt werden {value0}-{value1} von {value2}.",
"i18n:govoplan-audit.tenant_audit.f1a20817": "Mandantenaudit",
"i18n:govoplan-audit.tenant_context.f1a20818": "Mandantenkontext",
"i18n:govoplan-audit.tenant_level_administrative_history_showing_value0_value1_of_value2.f1a20819": "Administrative Historie des aktiven Mandanten, angezeigt werden {value0}-{value1} von {value2}.",
"i18n:govoplan-audit.time.f1a20820": "Zeit",
"i18n:govoplan-audit.value.f1a20821": "Wert"
}
};
+5
View File
@@ -0,0 +1,5 @@
export { default } from "./module";
export * from "./module";
export * from "./api/audit";
export { default as AdminAuditPanel } from "./features/audit/AdminAuditPanel";
export type { PlatformWebModule } from "@govoplan/core-webui";
+61
View File
@@ -0,0 +1,61 @@
import { createElement, lazy } from "react";
import { type AdminSectionsUiCapability, type PlatformWebModule } from "@govoplan/core-webui";
import { generatedTranslations } from "./i18n/generatedTranslations";
const AdminAuditPanel = lazy(() => import("./features/audit/AdminAuditPanel"));
const translations = {
en: generatedTranslations.en,
de: generatedTranslations.de
};
const auditAdminSections: AdminSectionsUiCapability = {
sections: [
{
id: "system-audit",
moduleId: "audit",
kind: "management",
surfaceId: "audit.admin.system",
label: "i18n:govoplan-audit.system_audit.f1a20815",
group: "SYSTEM",
order: 90,
allOf: ["system:audit:read"],
render: ({ settings, auth }) => createElement(AdminAuditPanel, {
settings,
auth,
systemMode: true
})
},
{
id: "tenant-audit",
moduleId: "audit",
kind: "management",
surfaceId: "audit.admin.tenant",
label: "i18n:govoplan-audit.tenant_audit.f1a20817",
group: "TENANT",
order: 100,
allOf: ["audit:read"],
render: ({ settings, auth }) => createElement(AdminAuditPanel, {
settings,
auth,
systemMode: false
})
}
]
};
export const auditModule: PlatformWebModule = {
id: "audit",
label: "Audit",
version: "0.1.8",
dependencies: ["access", "admin"],
viewSurfaces: [
{ id: "audit.admin.system", moduleId: "audit", kind: "section", label: "System audit", order: 90 },
{ id: "audit.admin.tenant", moduleId: "audit", kind: "section", label: "Tenant audit", order: 100 }
],
translations,
uiCapabilities: {
"admin.sections": auditAdminSections
}
};
export default auditModule;