1565 lines
48 KiB
Python
1565 lines
48 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping, Sequence
|
|
from datetime import UTC, datetime
|
|
import difflib
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.events import (
|
|
EventActorRef,
|
|
EventObjectRef,
|
|
EventTenantRef,
|
|
PlatformEvent,
|
|
emit_platform_event,
|
|
)
|
|
from govoplan_core.security.module_permissions import scopes_grant_compatible
|
|
from govoplan_wiki.backend.db.models import (
|
|
WikiComment,
|
|
WikiPage,
|
|
WikiPageRevision,
|
|
WikiSpace,
|
|
WikiSpaceHistory,
|
|
)
|
|
|
|
|
|
CAPABILITY_WIKI_REGISTRY = "wiki.registry"
|
|
READ_SCOPE = "wiki:page:read"
|
|
WRITE_SCOPE = "wiki:page:write"
|
|
COMMENT_SCOPE = "wiki:page:comment"
|
|
PUBLISH_SCOPE = "wiki:page:publish"
|
|
ADMIN_SCOPE = "wiki:space:admin"
|
|
_SLUG = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
|
|
_ACL_PREFIXES = frozenset(
|
|
{"account", "identity", "membership", "group", "role", "function_assignment"}
|
|
)
|
|
_VISIBILITIES = frozenset({"tenant", "restricted"})
|
|
_PUBLISH_MODES = frozenset({"publishers", "editors"})
|
|
|
|
|
|
class WikiStoreError(ValueError):
|
|
pass
|
|
|
|
|
|
class WikiNotFoundError(LookupError):
|
|
pass
|
|
|
|
|
|
class WikiConflictError(WikiStoreError):
|
|
pass
|
|
|
|
|
|
def create_space(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
space_id: str,
|
|
space_key: str,
|
|
title: str,
|
|
description: str,
|
|
visibility: str,
|
|
acl_tokens: Sequence[str],
|
|
publish_mode: str,
|
|
recorded_at: datetime,
|
|
change_reason: str,
|
|
idempotency_key: str,
|
|
registry: object | None = None,
|
|
) -> dict[str, object]:
|
|
_require_scope(principal, ADMIN_SCOPE)
|
|
tenant_id = _tenant(principal)
|
|
payload = {
|
|
"space_id": _bounded(space_id, "Space id", 255),
|
|
"space_key": _slug(space_key, "Space key", 120),
|
|
"title": _bounded(title, "Space title", 500),
|
|
"description": _optional(description, 20_000),
|
|
"visibility": _visibility(visibility),
|
|
"acl_tokens": _acl(acl_tokens),
|
|
"publish_mode": _choice(publish_mode, _PUBLISH_MODES, "publish mode"),
|
|
"recorded_at": _aware(recorded_at, "Recorded at"),
|
|
"change_reason": _bounded(change_reason, "Change reason", 1_000),
|
|
}
|
|
_restricted_requires_acl(payload["visibility"], payload["acl_tokens"])
|
|
key = _bounded(idempotency_key, "Idempotency key", 255)
|
|
request_hash = _sha(payload)
|
|
replay = _space_replay(session, tenant_id, key, request_hash)
|
|
if replay is not None:
|
|
return replay
|
|
if session.get(WikiSpace, payload["space_id"]) is not None:
|
|
raise WikiConflictError("A Wiki space with this identifier already exists.")
|
|
actor = _actor(principal)
|
|
row = WikiSpace(
|
|
id=payload["space_id"],
|
|
tenant_id=tenant_id,
|
|
space_key=payload["space_key"],
|
|
revision=1,
|
|
title=payload["title"],
|
|
description=payload["description"],
|
|
visibility=payload["visibility"],
|
|
acl_tokens=payload["acl_tokens"],
|
|
publish_mode=payload["publish_mode"],
|
|
created_by=actor,
|
|
updated_by=actor,
|
|
)
|
|
session.add(row)
|
|
session.flush()
|
|
snapshot = _space_payload(row)
|
|
_append_space_history(
|
|
session,
|
|
row,
|
|
"created",
|
|
payload["recorded_at"],
|
|
actor,
|
|
payload["change_reason"],
|
|
key,
|
|
request_hash,
|
|
snapshot,
|
|
)
|
|
_emit(
|
|
session,
|
|
registry,
|
|
row.tenant_id,
|
|
"wiki.space.created",
|
|
"wiki_space",
|
|
row.id,
|
|
row.title,
|
|
row.revision,
|
|
actor,
|
|
payload["recorded_at"],
|
|
)
|
|
return snapshot
|
|
|
|
|
|
def update_space(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
space_id: str,
|
|
expected_revision: int,
|
|
changes: Mapping[str, object],
|
|
recorded_at: datetime,
|
|
change_reason: str,
|
|
idempotency_key: str,
|
|
registry: object | None = None,
|
|
) -> dict[str, object]:
|
|
_require_scope(principal, ADMIN_SCOPE)
|
|
allowed = {"title", "description", "visibility", "acl_tokens", "publish_mode"}
|
|
unexpected = set(changes) - allowed
|
|
if unexpected:
|
|
raise WikiStoreError(
|
|
"Unsupported Wiki space changes: " + ", ".join(sorted(unexpected))
|
|
)
|
|
row = _space(session, principal, space_id, lock=True)
|
|
normalized: dict[str, object] = {}
|
|
if "title" in changes:
|
|
normalized["title"] = _bounded(changes["title"], "Space title", 500)
|
|
if "description" in changes:
|
|
normalized["description"] = _optional(changes["description"], 20_000)
|
|
if "visibility" in changes:
|
|
normalized["visibility"] = _visibility(changes["visibility"])
|
|
if "acl_tokens" in changes:
|
|
normalized["acl_tokens"] = _acl(changes["acl_tokens"])
|
|
if "publish_mode" in changes:
|
|
normalized["publish_mode"] = _choice(
|
|
changes["publish_mode"], _PUBLISH_MODES, "publish mode"
|
|
)
|
|
effective_visibility = str(normalized.get("visibility", row.visibility))
|
|
effective_acl = normalized.get("acl_tokens", row.acl_tokens)
|
|
_restricted_requires_acl(effective_visibility, effective_acl)
|
|
payload = {
|
|
"space_id": space_id,
|
|
"expected_revision": expected_revision,
|
|
"changes": normalized,
|
|
"recorded_at": _aware(recorded_at, "Recorded at"),
|
|
"change_reason": _bounded(change_reason, "Change reason", 1_000),
|
|
}
|
|
return _mutate_space(
|
|
session,
|
|
principal,
|
|
row,
|
|
expected_revision,
|
|
payload,
|
|
"updated",
|
|
idempotency_key,
|
|
registry,
|
|
)
|
|
|
|
|
|
def archive_space(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
space_id: str,
|
|
expected_revision: int,
|
|
recorded_at: datetime,
|
|
change_reason: str,
|
|
idempotency_key: str,
|
|
registry: object | None = None,
|
|
) -> dict[str, object]:
|
|
_require_scope(principal, ADMIN_SCOPE)
|
|
row = _space(session, principal, space_id, lock=True)
|
|
payload = {
|
|
"space_id": space_id,
|
|
"expected_revision": expected_revision,
|
|
"changes": {"archived_at": _aware(recorded_at, "Recorded at")},
|
|
"recorded_at": recorded_at,
|
|
"change_reason": _bounded(change_reason, "Change reason", 1_000),
|
|
}
|
|
return _mutate_space(
|
|
session,
|
|
principal,
|
|
row,
|
|
expected_revision,
|
|
payload,
|
|
"archived",
|
|
idempotency_key,
|
|
registry,
|
|
)
|
|
|
|
|
|
def list_spaces(
|
|
session: Session, principal: object, *, include_archived: bool = False
|
|
) -> tuple[dict[str, object], ...]:
|
|
_require_scope(principal, READ_SCOPE)
|
|
query = session.query(WikiSpace).filter(WikiSpace.tenant_id == _tenant(principal))
|
|
if not include_archived:
|
|
query = query.filter(WikiSpace.archived_at.is_(None))
|
|
return tuple(
|
|
_space_payload(row)
|
|
for row in query.order_by(WikiSpace.title.asc()).all()
|
|
if can_access_space(principal, row)
|
|
)
|
|
|
|
|
|
def get_space(
|
|
session: Session, principal: object, *, space_id: str
|
|
) -> dict[str, object] | None:
|
|
row = (
|
|
session.query(WikiSpace)
|
|
.filter(WikiSpace.tenant_id == _tenant(principal), WikiSpace.id == space_id)
|
|
.one_or_none()
|
|
)
|
|
return (
|
|
_space_payload(row)
|
|
if row is not None and can_access_space(principal, row)
|
|
else None
|
|
)
|
|
|
|
|
|
def create_page(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
page_id: str,
|
|
space_id: str,
|
|
parent_page_id: str | None,
|
|
slug: str,
|
|
title: str,
|
|
body: str,
|
|
summary: str,
|
|
inherits_access: bool,
|
|
visibility: str,
|
|
acl_tokens: Sequence[str],
|
|
labels: Sequence[str],
|
|
links: Sequence[Mapping[str, object]],
|
|
recorded_at: datetime,
|
|
change_reason: str,
|
|
idempotency_key: str,
|
|
registry: object | None = None,
|
|
) -> dict[str, object]:
|
|
_require_scope(principal, WRITE_SCOPE)
|
|
tenant_id = _tenant(principal)
|
|
space = _space(session, principal, space_id)
|
|
_require_space_write(principal, space)
|
|
if space.archived_at is not None:
|
|
raise WikiConflictError("Pages cannot be added to an archived Wiki space.")
|
|
parent = _parent(session, principal, space, parent_page_id)
|
|
clean_slug = _slug(slug, "Page slug", 160)
|
|
payload: dict[str, object] = {
|
|
"page_id": _bounded(page_id, "Page id", 255),
|
|
"space_id": space_id,
|
|
"parent_page_id": parent.id if parent else None,
|
|
"slug": clean_slug,
|
|
"path": f"{parent.path}/{clean_slug}" if parent else clean_slug,
|
|
"title": _bounded(title, "Page title", 500),
|
|
"body": _optional(body, 500_000),
|
|
"summary": _optional(summary, 10_000),
|
|
"inherits_access": bool(inherits_access),
|
|
"visibility": "inherit" if inherits_access else _visibility(visibility),
|
|
"acl_tokens": [] if inherits_access else _acl(acl_tokens),
|
|
"labels": _labels(labels),
|
|
"links": _links(links),
|
|
"recorded_at": _aware(recorded_at, "Recorded at"),
|
|
"change_reason": _bounded(change_reason, "Change reason", 1_000),
|
|
}
|
|
if not inherits_access:
|
|
_restricted_requires_acl(payload["visibility"], payload["acl_tokens"])
|
|
key = _bounded(idempotency_key, "Idempotency key", 255)
|
|
request_hash = _sha(payload)
|
|
replay = _page_replay(session, tenant_id, key, request_hash)
|
|
if replay is not None:
|
|
return replay
|
|
if session.get(WikiPage, payload["page_id"]) is not None:
|
|
raise WikiConflictError("A Wiki page with this identifier already exists.")
|
|
actor = _actor(principal)
|
|
row = WikiPage(
|
|
id=payload["page_id"],
|
|
tenant_id=tenant_id,
|
|
space_id=space.id,
|
|
parent_page_id=payload["parent_page_id"],
|
|
slug=payload["slug"],
|
|
path=payload["path"],
|
|
revision=1,
|
|
published_revision=None,
|
|
state="draft",
|
|
title=payload["title"],
|
|
body=payload["body"],
|
|
summary=payload["summary"],
|
|
visibility=payload["visibility"],
|
|
inherits_access=payload["inherits_access"],
|
|
acl_tokens=payload["acl_tokens"],
|
|
labels=payload["labels"],
|
|
links=payload["links"],
|
|
redirect_page_id=None,
|
|
search_text=_search_text(payload),
|
|
created_by=actor,
|
|
updated_by=actor,
|
|
)
|
|
session.add(row)
|
|
session.flush()
|
|
snapshot = _page_payload(row, space=space)
|
|
_append_page_revision(
|
|
session,
|
|
row,
|
|
"created",
|
|
payload["recorded_at"],
|
|
actor,
|
|
payload["change_reason"],
|
|
key,
|
|
request_hash,
|
|
snapshot,
|
|
)
|
|
_emit(
|
|
session,
|
|
registry,
|
|
tenant_id,
|
|
"wiki.page.created",
|
|
"wiki_page",
|
|
row.id,
|
|
row.title,
|
|
row.revision,
|
|
actor,
|
|
payload["recorded_at"],
|
|
)
|
|
return snapshot
|
|
|
|
|
|
def update_page(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
page_id: str,
|
|
expected_revision: int,
|
|
changes: Mapping[str, object],
|
|
recorded_at: datetime,
|
|
change_reason: str,
|
|
idempotency_key: str,
|
|
registry: object | None = None,
|
|
) -> dict[str, object]:
|
|
_require_scope(principal, WRITE_SCOPE)
|
|
row, space = _page_and_space(session, principal, page_id, lock=True, govern=True)
|
|
if space.archived_at is not None or row.state == "archived":
|
|
raise WikiConflictError("Archived Wiki content cannot be revised.")
|
|
allowed = {
|
|
"title",
|
|
"body",
|
|
"summary",
|
|
"slug",
|
|
"parent_page_id",
|
|
"inherits_access",
|
|
"visibility",
|
|
"acl_tokens",
|
|
"labels",
|
|
"links",
|
|
"redirect_page_id",
|
|
}
|
|
unexpected = set(changes) - allowed
|
|
if unexpected:
|
|
raise WikiStoreError(
|
|
"Unsupported Wiki page changes: " + ", ".join(sorted(unexpected))
|
|
)
|
|
normalized: dict[str, object] = {}
|
|
for name, maximum in (("title", 500), ("body", 500_000), ("summary", 10_000)):
|
|
if name in changes:
|
|
normalized[name] = (
|
|
_bounded(changes[name], "Page title", maximum)
|
|
if name == "title"
|
|
else _optional(changes[name], maximum)
|
|
)
|
|
if "slug" in changes:
|
|
normalized["slug"] = _slug(changes["slug"], "Page slug", 160)
|
|
if "labels" in changes:
|
|
normalized["labels"] = _labels(changes["labels"])
|
|
if "links" in changes:
|
|
normalized["links"] = _links(changes["links"])
|
|
if "inherits_access" in changes:
|
|
normalized["inherits_access"] = bool(changes["inherits_access"])
|
|
inherits = bool(normalized.get("inherits_access", row.inherits_access))
|
|
if inherits:
|
|
normalized["visibility"] = "inherit"
|
|
normalized["acl_tokens"] = []
|
|
else:
|
|
if "visibility" in changes:
|
|
normalized["visibility"] = _visibility(changes["visibility"])
|
|
elif row.visibility == "inherit":
|
|
normalized["visibility"] = "tenant"
|
|
if "acl_tokens" in changes:
|
|
normalized["acl_tokens"] = _acl(changes["acl_tokens"])
|
|
effective_visibility = str(normalized.get("visibility", row.visibility))
|
|
effective_acl = normalized.get("acl_tokens", row.acl_tokens)
|
|
_restricted_requires_acl(effective_visibility, effective_acl)
|
|
new_parent_id = changes.get("parent_page_id", row.parent_page_id)
|
|
parent = _parent(
|
|
session, principal, space, str(new_parent_id) if new_parent_id else None
|
|
)
|
|
if parent is not None and parent.id == row.id:
|
|
raise WikiStoreError("A page cannot be its own parent.")
|
|
if ("parent_page_id" in changes or "slug" in changes) and session.query(
|
|
WikiPage.id
|
|
).filter(
|
|
WikiPage.tenant_id == row.tenant_id, WikiPage.parent_page_id == row.id
|
|
).first():
|
|
raise WikiConflictError(
|
|
"Move or rename child pages before changing this page path."
|
|
)
|
|
normalized["parent_page_id"] = parent.id if parent else None
|
|
next_slug = str(normalized.get("slug", row.slug))
|
|
normalized["path"] = f"{parent.path}/{next_slug}" if parent else next_slug
|
|
if "redirect_page_id" in changes:
|
|
target = changes["redirect_page_id"]
|
|
if target is not None:
|
|
target_row = (
|
|
session.query(WikiPage)
|
|
.filter(
|
|
WikiPage.tenant_id == row.tenant_id,
|
|
WikiPage.space_id == row.space_id,
|
|
WikiPage.id == str(target),
|
|
)
|
|
.one_or_none()
|
|
)
|
|
if target_row is None or target_row.id == row.id:
|
|
raise WikiStoreError(
|
|
"Redirect target must be another page in the same space."
|
|
)
|
|
normalized["redirect_page_id"] = target_row.id
|
|
else:
|
|
normalized["redirect_page_id"] = None
|
|
# Every content update is a draft. The last published revision remains visible
|
|
# until a separate publish action appends the next immutable revision.
|
|
normalized["state"] = "draft"
|
|
payload = {
|
|
"page_id": page_id,
|
|
"expected_revision": expected_revision,
|
|
"changes": normalized,
|
|
"recorded_at": _aware(recorded_at, "Recorded at"),
|
|
"change_reason": _bounded(change_reason, "Change reason", 1_000),
|
|
}
|
|
return _mutate_page(
|
|
session,
|
|
principal,
|
|
row,
|
|
space,
|
|
expected_revision,
|
|
payload,
|
|
"revised",
|
|
idempotency_key,
|
|
registry,
|
|
)
|
|
|
|
|
|
def publish_page(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
page_id: str,
|
|
expected_revision: int,
|
|
recorded_at: datetime,
|
|
change_reason: str,
|
|
idempotency_key: str,
|
|
registry: object | None = None,
|
|
) -> dict[str, object]:
|
|
row, space = _page_and_space(session, principal, page_id, lock=True, govern=True)
|
|
if space.publish_mode == "editors":
|
|
_require_any_scope(principal, WRITE_SCOPE, PUBLISH_SCOPE, ADMIN_SCOPE)
|
|
else:
|
|
_require_any_scope(principal, PUBLISH_SCOPE, ADMIN_SCOPE)
|
|
if row.state == "archived" or space.archived_at is not None:
|
|
raise WikiConflictError("Archived Wiki content cannot be published.")
|
|
next_revision = expected_revision + 1
|
|
payload = {
|
|
"page_id": page_id,
|
|
"expected_revision": expected_revision,
|
|
"changes": {
|
|
"state": "redirected" if row.redirect_page_id else "published",
|
|
"published_revision": next_revision,
|
|
"published_at": _aware(recorded_at, "Recorded at"),
|
|
"archived_at": None,
|
|
},
|
|
"recorded_at": recorded_at,
|
|
"change_reason": _bounded(change_reason, "Change reason", 1_000),
|
|
}
|
|
return _mutate_page(
|
|
session,
|
|
principal,
|
|
row,
|
|
space,
|
|
expected_revision,
|
|
payload,
|
|
"published",
|
|
idempotency_key,
|
|
registry,
|
|
)
|
|
|
|
|
|
def archive_page(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
page_id: str,
|
|
expected_revision: int,
|
|
recorded_at: datetime,
|
|
change_reason: str,
|
|
idempotency_key: str,
|
|
registry: object | None = None,
|
|
) -> dict[str, object]:
|
|
_require_any_scope(principal, PUBLISH_SCOPE, ADMIN_SCOPE)
|
|
row, space = _page_and_space(session, principal, page_id, lock=True, govern=True)
|
|
if (
|
|
session.query(WikiPage.id)
|
|
.filter(
|
|
WikiPage.tenant_id == row.tenant_id,
|
|
WikiPage.parent_page_id == row.id,
|
|
WikiPage.state != "archived",
|
|
)
|
|
.first()
|
|
):
|
|
raise WikiConflictError("Archive or move active child pages first.")
|
|
payload = {
|
|
"page_id": page_id,
|
|
"expected_revision": expected_revision,
|
|
"changes": {
|
|
"state": "archived",
|
|
"archived_at": _aware(recorded_at, "Recorded at"),
|
|
},
|
|
"recorded_at": recorded_at,
|
|
"change_reason": _bounded(change_reason, "Change reason", 1_000),
|
|
}
|
|
return _mutate_page(
|
|
session,
|
|
principal,
|
|
row,
|
|
space,
|
|
expected_revision,
|
|
payload,
|
|
"archived",
|
|
idempotency_key,
|
|
registry,
|
|
)
|
|
|
|
|
|
def get_page(
|
|
session: Session, principal: object, *, page_id: str
|
|
) -> dict[str, object] | None:
|
|
row = (
|
|
session.query(WikiPage)
|
|
.filter(WikiPage.tenant_id == _tenant(principal), WikiPage.id == page_id)
|
|
.one_or_none()
|
|
)
|
|
if row is None:
|
|
return None
|
|
space = (
|
|
session.query(WikiSpace)
|
|
.filter(WikiSpace.tenant_id == row.tenant_id, WikiSpace.id == row.space_id)
|
|
.one()
|
|
)
|
|
if not can_read_page(session, principal, row, space=space):
|
|
return None
|
|
if not _editor(principal) and row.published_revision != row.revision:
|
|
revision = _revision(session, row, row.published_revision)
|
|
return _published_payload(revision, space) if revision is not None else None
|
|
return _page_payload(row, space=space)
|
|
|
|
|
|
def list_pages(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
space_id: str | None = None,
|
|
parent_page_id: str | None = None,
|
|
states: Sequence[str] = (),
|
|
query: str = "",
|
|
include_archived: bool = False,
|
|
offset: int = 0,
|
|
limit: int = 100,
|
|
) -> tuple[tuple[dict[str, object], ...], int]:
|
|
_require_scope(principal, READ_SCOPE)
|
|
if offset < 0 or not 1 <= limit <= 200:
|
|
raise WikiStoreError(
|
|
"Page list offset must be non-negative and limit between 1 and 200."
|
|
)
|
|
statement = session.query(WikiPage).filter(WikiPage.tenant_id == _tenant(principal))
|
|
if space_id:
|
|
statement = statement.filter(WikiPage.space_id == space_id)
|
|
if parent_page_id is not None:
|
|
statement = statement.filter(WikiPage.parent_page_id == parent_page_id)
|
|
if states:
|
|
statement = statement.filter(WikiPage.state.in_(tuple(states)))
|
|
if not include_archived:
|
|
statement = statement.filter(WikiPage.state != "archived")
|
|
clean_query = str(query or "").strip().casefold()
|
|
if clean_query:
|
|
statement = statement.filter(WikiPage.search_text.contains(clean_query))
|
|
spaces = {
|
|
row.id: row
|
|
for row in session.query(WikiSpace)
|
|
.filter(WikiSpace.tenant_id == _tenant(principal))
|
|
.all()
|
|
}
|
|
accessible = []
|
|
for row in statement.order_by(WikiPage.path.asc()).all():
|
|
space = spaces.get(row.space_id)
|
|
if space is None or not can_read_page(session, principal, row, space=space):
|
|
continue
|
|
if not _editor(principal) and row.published_revision != row.revision:
|
|
published = _revision(session, row, row.published_revision)
|
|
if published is not None:
|
|
accessible.append(_published_payload(published, space))
|
|
else:
|
|
accessible.append(_page_payload(row, space=space))
|
|
return tuple(accessible[offset : offset + limit]), len(accessible)
|
|
|
|
|
|
def page_revisions(
|
|
session: Session, principal: object, *, page_id: str
|
|
) -> tuple[dict[str, object], ...]:
|
|
row, _space_row = _page_and_space(session, principal, page_id)
|
|
if not _editor(principal):
|
|
raise PermissionError("Revision history requires Wiki edit access.")
|
|
rows = (
|
|
session.query(WikiPageRevision)
|
|
.filter(
|
|
WikiPageRevision.tenant_id == row.tenant_id,
|
|
WikiPageRevision.page_id == row.id,
|
|
)
|
|
.order_by(WikiPageRevision.revision.desc())
|
|
.limit(500)
|
|
.all()
|
|
)
|
|
return tuple(
|
|
{
|
|
"revision": item.revision,
|
|
"event_type": item.event_type,
|
|
"recorded_at": _iso(item.recorded_at),
|
|
"actor_id": item.actor_id,
|
|
"change_reason": item.change_reason,
|
|
"snapshot": dict(item.snapshot),
|
|
}
|
|
for item in rows
|
|
)
|
|
|
|
|
|
def compare_revisions(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
page_id: str,
|
|
from_revision: int,
|
|
to_revision: int,
|
|
) -> dict[str, object]:
|
|
row, _space_row = _page_and_space(session, principal, page_id)
|
|
if not _editor(principal):
|
|
raise PermissionError("Revision comparison requires Wiki edit access.")
|
|
before = _revision(session, row, from_revision)
|
|
after = _revision(session, row, to_revision)
|
|
if before is None or after is None:
|
|
raise WikiNotFoundError("Wiki page revision not found.")
|
|
before_body = str(before.snapshot.get("body") or "")
|
|
after_body = str(after.snapshot.get("body") or "")
|
|
lines = list(
|
|
difflib.unified_diff(
|
|
before_body.splitlines(),
|
|
after_body.splitlines(),
|
|
fromfile=f"revision-{from_revision}",
|
|
tofile=f"revision-{to_revision}",
|
|
lineterm="",
|
|
)
|
|
)
|
|
joined = "\n".join(lines)
|
|
truncated = len(joined) > 200_000
|
|
return {
|
|
"page_id": page_id,
|
|
"from_revision": from_revision,
|
|
"to_revision": to_revision,
|
|
"diff": joined[:200_000],
|
|
"truncated": truncated,
|
|
}
|
|
|
|
|
|
def add_comment(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
page_id: str,
|
|
expected_revision: int,
|
|
comment_id: str,
|
|
body: str,
|
|
recorded_at: datetime,
|
|
idempotency_key: str,
|
|
registry: object | None = None,
|
|
) -> dict[str, object]:
|
|
_require_any_scope(principal, COMMENT_SCOPE, WRITE_SCOPE, ADMIN_SCOPE)
|
|
row, _space_row = _page_and_space(session, principal, page_id, govern=True)
|
|
payload = {
|
|
"page_id": page_id,
|
|
"expected_revision": expected_revision,
|
|
"comment_id": _bounded(comment_id, "Comment id", 255),
|
|
"body": _bounded(body, "Comment body", 20_000),
|
|
"recorded_at": _aware(recorded_at, "Recorded at"),
|
|
}
|
|
key = _bounded(idempotency_key, "Idempotency key", 255)
|
|
request_hash = _sha(payload)
|
|
replay = (
|
|
session.query(WikiComment)
|
|
.filter(
|
|
WikiComment.tenant_id == row.tenant_id, WikiComment.idempotency_key == key
|
|
)
|
|
.one_or_none()
|
|
)
|
|
if replay is not None:
|
|
if replay.request_sha256 != request_hash:
|
|
raise WikiConflictError(
|
|
"Idempotency key was already used with a different comment request."
|
|
)
|
|
return _comment_payload(replay)
|
|
if row.revision != expected_revision:
|
|
raise WikiConflictError(
|
|
f"Wiki page revision conflict: expected {expected_revision}, current {row.revision}."
|
|
)
|
|
comment = WikiComment(
|
|
tenant_id=row.tenant_id,
|
|
page_id=row.id,
|
|
comment_id=payload["comment_id"],
|
|
page_revision=row.revision,
|
|
body=payload["body"],
|
|
created_by=_actor(principal),
|
|
recorded_at=payload["recorded_at"],
|
|
idempotency_key=key,
|
|
request_sha256=request_hash,
|
|
)
|
|
session.add(comment)
|
|
session.flush()
|
|
_emit(
|
|
session,
|
|
registry,
|
|
row.tenant_id,
|
|
"wiki.page.commented",
|
|
"wiki_page",
|
|
row.id,
|
|
row.title,
|
|
row.revision,
|
|
_actor(principal),
|
|
payload["recorded_at"],
|
|
)
|
|
return _comment_payload(comment)
|
|
|
|
|
|
def list_comments(
|
|
session: Session, principal: object, *, page_id: str
|
|
) -> tuple[dict[str, object], ...]:
|
|
row, _space_row = _page_and_space(session, principal, page_id)
|
|
return tuple(
|
|
_comment_payload(item)
|
|
for item in session.query(WikiComment)
|
|
.filter(WikiComment.tenant_id == row.tenant_id, WikiComment.page_id == row.id)
|
|
.order_by(WikiComment.recorded_at.asc())
|
|
.limit(500)
|
|
.all()
|
|
)
|
|
|
|
|
|
def can_access_space(principal: object, row: WikiSpace) -> bool:
|
|
return (
|
|
row.tenant_id == _tenant(principal)
|
|
and _has_scope(principal, READ_SCOPE)
|
|
and (
|
|
_has_scope(principal, ADMIN_SCOPE)
|
|
or row.visibility == "tenant"
|
|
or bool(set(row.acl_tokens or ()) & set(_principal_tokens(principal)))
|
|
)
|
|
)
|
|
|
|
|
|
def can_read_page(
|
|
session: Session,
|
|
principal: object,
|
|
row: WikiPage,
|
|
*,
|
|
space: WikiSpace | None = None,
|
|
) -> bool:
|
|
if row.tenant_id != _tenant(principal) or not _has_scope(principal, READ_SCOPE):
|
|
return False
|
|
space = (
|
|
space
|
|
or session.query(WikiSpace)
|
|
.filter(WikiSpace.tenant_id == row.tenant_id, WikiSpace.id == row.space_id)
|
|
.one_or_none()
|
|
)
|
|
if space is None or not _access(
|
|
row.inherits_access, row.visibility, row.acl_tokens, space, principal
|
|
):
|
|
return False
|
|
if _editor(principal):
|
|
return True
|
|
return row.published_revision is not None and row.state != "archived"
|
|
|
|
|
|
class SqlWikiRegistry:
|
|
def resolve_page(
|
|
self, session: object, principal: object, *, page_id: str
|
|
) -> dict[str, object] | None:
|
|
if not isinstance(session, Session):
|
|
raise TypeError("Wiki registry requires a SQLAlchemy session.")
|
|
page = get_page(session, principal, page_id=page_id)
|
|
if page is None:
|
|
return None
|
|
return {
|
|
"page_id": page["page_id"],
|
|
"space_id": page["space_id"],
|
|
"title": page["title"],
|
|
"state": page["state"],
|
|
"revision": page["revision"],
|
|
"url": f"/wiki?pageId={page['page_id']}",
|
|
}
|
|
|
|
|
|
def integration_availability(registry: object | None) -> dict[str, object]:
|
|
active = _active_modules(registry)
|
|
return {
|
|
"files": "files" in active,
|
|
"dms": "dms" in active,
|
|
"search": "search" in active,
|
|
"templates": "templates" in active,
|
|
"workflow_engine": "workflow_engine" in active,
|
|
"projects": "projects" in active,
|
|
"cases": "cases" in active,
|
|
"notifications": "notifications" in active,
|
|
"consequences": {
|
|
"files": "Attachments remain governed references; Wiki never stores binary content.",
|
|
"search": "Published pages remain browsable in Wiki but are not projected into global Search.",
|
|
"connectors": "External Wiki transport and migration are unavailable without a connector provider.",
|
|
},
|
|
}
|
|
|
|
|
|
def _mutate_space(
|
|
session: Session,
|
|
principal: object,
|
|
row: WikiSpace,
|
|
expected_revision: int,
|
|
payload: dict[str, object],
|
|
event_type: str,
|
|
idempotency_key: str,
|
|
registry: object | None,
|
|
) -> dict[str, object]:
|
|
key = _bounded(idempotency_key, "Idempotency key", 255)
|
|
request_hash = _sha(payload)
|
|
replay = _space_replay(session, row.tenant_id, key, request_hash)
|
|
if replay is not None:
|
|
return replay
|
|
if event_type == "archived" and row.archived_at is not None:
|
|
raise WikiConflictError("Wiki space is already archived.")
|
|
if row.revision != expected_revision:
|
|
raise WikiConflictError(
|
|
f"Wiki space revision conflict: expected {expected_revision}, current {row.revision}."
|
|
)
|
|
for name, value in dict(payload["changes"]).items():
|
|
setattr(row, name, value)
|
|
row.revision += 1
|
|
row.updated_by = _actor(principal)
|
|
session.flush()
|
|
snapshot = _space_payload(row)
|
|
_append_space_history(
|
|
session,
|
|
row,
|
|
event_type,
|
|
payload["recorded_at"],
|
|
row.updated_by,
|
|
payload["change_reason"],
|
|
key,
|
|
request_hash,
|
|
snapshot,
|
|
)
|
|
_emit(
|
|
session,
|
|
registry,
|
|
row.tenant_id,
|
|
f"wiki.space.{event_type}",
|
|
"wiki_space",
|
|
row.id,
|
|
row.title,
|
|
row.revision,
|
|
row.updated_by,
|
|
payload["recorded_at"],
|
|
)
|
|
return snapshot
|
|
|
|
|
|
def _mutate_page(
|
|
session: Session,
|
|
principal: object,
|
|
row: WikiPage,
|
|
space: WikiSpace,
|
|
expected_revision: int,
|
|
payload: dict[str, object],
|
|
event_type: str,
|
|
idempotency_key: str,
|
|
registry: object | None,
|
|
) -> dict[str, object]:
|
|
key = _bounded(idempotency_key, "Idempotency key", 255)
|
|
request_hash = _sha(payload)
|
|
replay = _page_replay(session, row.tenant_id, key, request_hash)
|
|
if replay is not None:
|
|
return replay
|
|
if event_type == "archived" and row.state == "archived":
|
|
raise WikiConflictError("Wiki page is already archived.")
|
|
if event_type == "published" and row.state in {"published", "redirected"}:
|
|
raise WikiConflictError("Wiki page revision is already published.")
|
|
if row.revision != expected_revision:
|
|
raise WikiConflictError(
|
|
f"Wiki page revision conflict: expected {expected_revision}, current {row.revision}."
|
|
)
|
|
for name, value in dict(payload["changes"]).items():
|
|
setattr(row, name, value)
|
|
row.revision += 1
|
|
row.updated_by = _actor(principal)
|
|
row.search_text = _search_text(
|
|
{
|
|
"title": row.title,
|
|
"body": row.body,
|
|
"summary": row.summary,
|
|
"labels": row.labels,
|
|
"links": row.links,
|
|
}
|
|
)
|
|
session.flush()
|
|
snapshot = _page_payload(row, space=space)
|
|
_append_page_revision(
|
|
session,
|
|
row,
|
|
event_type,
|
|
payload["recorded_at"],
|
|
row.updated_by,
|
|
payload["change_reason"],
|
|
key,
|
|
request_hash,
|
|
snapshot,
|
|
)
|
|
_emit(
|
|
session,
|
|
registry,
|
|
row.tenant_id,
|
|
f"wiki.page.{event_type}",
|
|
"wiki_page",
|
|
row.id,
|
|
row.title,
|
|
row.revision,
|
|
row.updated_by,
|
|
payload["recorded_at"],
|
|
)
|
|
return snapshot
|
|
|
|
|
|
def _space(
|
|
session: Session, principal: object, space_id: str, *, lock: bool = False
|
|
) -> WikiSpace:
|
|
query = session.query(WikiSpace).filter(
|
|
WikiSpace.tenant_id == _tenant(principal), WikiSpace.id == space_id
|
|
)
|
|
row = query.with_for_update().one_or_none() if lock else query.one_or_none()
|
|
if row is None:
|
|
raise WikiNotFoundError("Wiki space not found.")
|
|
return row
|
|
|
|
|
|
def _page_and_space(
|
|
session: Session,
|
|
principal: object,
|
|
page_id: str,
|
|
*,
|
|
lock: bool = False,
|
|
edit: bool = False,
|
|
govern: bool = False,
|
|
) -> tuple[WikiPage, WikiSpace]:
|
|
query = session.query(WikiPage).filter(
|
|
WikiPage.tenant_id == _tenant(principal), WikiPage.id == page_id
|
|
)
|
|
row = query.with_for_update().one_or_none() if lock else query.one_or_none()
|
|
if row is None:
|
|
raise WikiNotFoundError("Wiki page not found.")
|
|
space = (
|
|
session.query(WikiSpace)
|
|
.filter(WikiSpace.tenant_id == row.tenant_id, WikiSpace.id == row.space_id)
|
|
.one()
|
|
)
|
|
if edit:
|
|
_require_space_write(principal, space)
|
|
if not _access(
|
|
row.inherits_access, row.visibility, row.acl_tokens, space, principal
|
|
):
|
|
raise PermissionError("Wiki page access is restricted.")
|
|
elif govern:
|
|
if not _access(
|
|
row.inherits_access, row.visibility, row.acl_tokens, space, principal
|
|
):
|
|
raise PermissionError("Wiki page access is restricted.")
|
|
elif not can_read_page(session, principal, row, space=space):
|
|
raise WikiNotFoundError("Wiki page not found.")
|
|
return row, space
|
|
|
|
|
|
def _parent(
|
|
session: Session, principal: object, space: WikiSpace, page_id: str | None
|
|
) -> WikiPage | None:
|
|
if page_id is None:
|
|
return None
|
|
row = (
|
|
session.query(WikiPage)
|
|
.filter(
|
|
WikiPage.tenant_id == space.tenant_id,
|
|
WikiPage.space_id == space.id,
|
|
WikiPage.id == page_id,
|
|
)
|
|
.one_or_none()
|
|
)
|
|
if row is None or not can_read_page(session, principal, row, space=space):
|
|
raise WikiStoreError("Parent page must be accessible in the same space.")
|
|
if row.state == "archived":
|
|
raise WikiStoreError("Archived pages cannot be parents.")
|
|
return row
|
|
|
|
|
|
def _require_space_write(principal: object, space: WikiSpace) -> None:
|
|
_require_scope(principal, WRITE_SCOPE)
|
|
if not _access(True, "inherit", (), space, principal):
|
|
raise PermissionError("Wiki space is not writable by this principal.")
|
|
|
|
|
|
def _access(
|
|
inherits: bool,
|
|
visibility: str,
|
|
tokens: Sequence[str],
|
|
space: WikiSpace,
|
|
principal: object,
|
|
) -> bool:
|
|
if _has_scope(principal, ADMIN_SCOPE):
|
|
return True
|
|
effective_visibility = space.visibility if inherits else visibility
|
|
effective_tokens = space.acl_tokens if inherits else tokens
|
|
return effective_visibility == "tenant" or bool(
|
|
set(effective_tokens or ()) & set(_principal_tokens(principal))
|
|
)
|
|
|
|
|
|
def _editor(principal: object) -> bool:
|
|
return _has_any_scope(principal, WRITE_SCOPE, PUBLISH_SCOPE, ADMIN_SCOPE)
|
|
|
|
|
|
def _page_payload(row: WikiPage, *, space: WikiSpace) -> dict[str, object]:
|
|
visibility = space.visibility if row.inherits_access else row.visibility
|
|
acl_tokens = (
|
|
list(space.acl_tokens or ())
|
|
if row.inherits_access
|
|
else list(row.acl_tokens or ())
|
|
)
|
|
return {
|
|
"page_id": row.id,
|
|
"tenant_id": row.tenant_id,
|
|
"space_id": row.space_id,
|
|
"parent_page_id": row.parent_page_id,
|
|
"slug": row.slug,
|
|
"path": row.path,
|
|
"revision": row.revision,
|
|
"published_revision": row.published_revision,
|
|
"state": row.state,
|
|
"title": row.title,
|
|
"body": row.body,
|
|
"summary": row.summary,
|
|
"inherits_access": row.inherits_access,
|
|
"visibility": row.visibility,
|
|
"effective_visibility": visibility,
|
|
"acl_tokens": list(row.acl_tokens or ()),
|
|
"effective_acl_tokens": acl_tokens,
|
|
"labels": list(row.labels or ()),
|
|
"links": list(row.links or ()),
|
|
"redirect_page_id": row.redirect_page_id,
|
|
"published_at": _iso(row.published_at),
|
|
"archived_at": _iso(row.archived_at),
|
|
"created_by": row.created_by,
|
|
"updated_by": row.updated_by,
|
|
"created_at": _iso(row.created_at),
|
|
"updated_at": _iso(row.updated_at),
|
|
}
|
|
|
|
|
|
def _published_payload(
|
|
revision: WikiPageRevision, space: WikiSpace
|
|
) -> dict[str, object]:
|
|
payload = dict(revision.snapshot)
|
|
if bool(payload.get("inherits_access")):
|
|
payload["effective_visibility"] = space.visibility
|
|
payload["effective_acl_tokens"] = list(space.acl_tokens or ())
|
|
return payload
|
|
|
|
|
|
def _space_payload(row: WikiSpace) -> dict[str, object]:
|
|
return {
|
|
"space_id": row.id,
|
|
"tenant_id": row.tenant_id,
|
|
"space_key": row.space_key,
|
|
"revision": row.revision,
|
|
"title": row.title,
|
|
"description": row.description,
|
|
"visibility": row.visibility,
|
|
"acl_tokens": list(row.acl_tokens or ()),
|
|
"publish_mode": row.publish_mode,
|
|
"archived_at": _iso(row.archived_at),
|
|
"created_by": row.created_by,
|
|
"updated_by": row.updated_by,
|
|
"created_at": _iso(row.created_at),
|
|
"updated_at": _iso(row.updated_at),
|
|
}
|
|
|
|
|
|
def _comment_payload(row: WikiComment) -> dict[str, object]:
|
|
return {
|
|
"comment_id": row.comment_id,
|
|
"page_id": row.page_id,
|
|
"page_revision": row.page_revision,
|
|
"body": row.body,
|
|
"created_by": row.created_by,
|
|
"recorded_at": _iso(row.recorded_at),
|
|
}
|
|
|
|
|
|
def _append_space_history(
|
|
session: Session,
|
|
row: WikiSpace,
|
|
event: str,
|
|
at: datetime,
|
|
actor: str | None,
|
|
reason: str,
|
|
key: str,
|
|
request_hash: str,
|
|
snapshot: dict[str, object],
|
|
) -> None:
|
|
session.add(
|
|
WikiSpaceHistory(
|
|
tenant_id=row.tenant_id,
|
|
space_id=row.id,
|
|
revision=row.revision,
|
|
event_type=event,
|
|
recorded_at=at,
|
|
actor_id=actor,
|
|
change_reason=reason,
|
|
idempotency_key=key,
|
|
request_sha256=request_hash,
|
|
snapshot=snapshot,
|
|
)
|
|
)
|
|
session.flush()
|
|
|
|
|
|
def _append_page_revision(
|
|
session: Session,
|
|
row: WikiPage,
|
|
event: str,
|
|
at: datetime,
|
|
actor: str | None,
|
|
reason: str,
|
|
key: str,
|
|
request_hash: str,
|
|
snapshot: dict[str, object],
|
|
) -> None:
|
|
session.add(
|
|
WikiPageRevision(
|
|
tenant_id=row.tenant_id,
|
|
page_id=row.id,
|
|
revision=row.revision,
|
|
event_type=event,
|
|
recorded_at=at,
|
|
actor_id=actor,
|
|
change_reason=reason,
|
|
idempotency_key=key,
|
|
request_sha256=request_hash,
|
|
snapshot=snapshot,
|
|
)
|
|
)
|
|
session.flush()
|
|
|
|
|
|
def _space_replay(
|
|
session: Session, tenant_id: str, key: str, request_hash: str
|
|
) -> dict[str, object] | None:
|
|
row = (
|
|
session.query(WikiSpaceHistory)
|
|
.filter(
|
|
WikiSpaceHistory.tenant_id == tenant_id,
|
|
WikiSpaceHistory.idempotency_key == key,
|
|
)
|
|
.one_or_none()
|
|
)
|
|
if row is None:
|
|
return None
|
|
if row.request_sha256 != request_hash:
|
|
raise WikiConflictError(
|
|
"Idempotency key was already used with a different space request."
|
|
)
|
|
return dict(row.snapshot)
|
|
|
|
|
|
def _page_replay(
|
|
session: Session, tenant_id: str, key: str, request_hash: str
|
|
) -> dict[str, object] | None:
|
|
row = (
|
|
session.query(WikiPageRevision)
|
|
.filter(
|
|
WikiPageRevision.tenant_id == tenant_id,
|
|
WikiPageRevision.idempotency_key == key,
|
|
)
|
|
.one_or_none()
|
|
)
|
|
if row is None:
|
|
return None
|
|
if row.request_sha256 != request_hash:
|
|
raise WikiConflictError(
|
|
"Idempotency key was already used with a different page request."
|
|
)
|
|
return dict(row.snapshot)
|
|
|
|
|
|
def _revision(
|
|
session: Session, row: WikiPage, revision: int | None
|
|
) -> WikiPageRevision | None:
|
|
if revision is None:
|
|
return None
|
|
return (
|
|
session.query(WikiPageRevision)
|
|
.filter(
|
|
WikiPageRevision.tenant_id == row.tenant_id,
|
|
WikiPageRevision.page_id == row.id,
|
|
WikiPageRevision.revision == revision,
|
|
)
|
|
.one_or_none()
|
|
)
|
|
|
|
|
|
def _labels(values: object) -> list[str]:
|
|
if not isinstance(values, Sequence) or isinstance(values, (str, bytes)):
|
|
raise WikiStoreError("Page labels must be a list.")
|
|
labels = tuple(
|
|
dict.fromkeys(_bounded(item, "Page label", 120).casefold() for item in values)
|
|
)
|
|
if len(labels) > 50:
|
|
raise WikiStoreError("A Wiki page supports at most 50 labels.")
|
|
return list(labels)
|
|
|
|
|
|
def _links(values: object) -> list[dict[str, object]]:
|
|
if not isinstance(values, Sequence) or isinstance(values, (str, bytes)):
|
|
raise WikiStoreError("Page links must be a list.")
|
|
if len(values) > 200:
|
|
raise WikiStoreError("A Wiki page supports at most 200 links.")
|
|
links: list[dict[str, object]] = []
|
|
seen: set[str] = set()
|
|
for raw in values:
|
|
if not isinstance(raw, Mapping):
|
|
raise WikiStoreError("Each Wiki link must be an object.")
|
|
link_id = _bounded(raw.get("link_id"), "Link id", 255)
|
|
if link_id in seen:
|
|
raise WikiStoreError("Wiki link identifiers must be unique per page.")
|
|
seen.add(link_id)
|
|
url = str(raw.get("url") or "").strip()
|
|
if url and not (
|
|
url.startswith("/")
|
|
or url.startswith("https://")
|
|
or url.startswith("http://")
|
|
):
|
|
raise WikiStoreError("Wiki link URLs must be local or HTTP(S).")
|
|
owner = _bounded(raw.get("owner_module"), "Link owner module", 120)
|
|
item = {
|
|
"link_id": link_id,
|
|
"kind": _choice(
|
|
raw.get("kind", "reference"),
|
|
{"attachment", "related", "reference", "external"},
|
|
"link kind",
|
|
),
|
|
"owner_module": owner,
|
|
"resource_type": _bounded(
|
|
raw.get("resource_type"), "Link resource type", 120
|
|
),
|
|
"resource_id": _bounded(raw.get("resource_id"), "Link resource id", 255),
|
|
"label": _optional(raw.get("label"), 500),
|
|
"url": url or None,
|
|
"external_system": _optional(raw.get("external_system"), 120),
|
|
"external_id": _optional(raw.get("external_id"), 255),
|
|
"provenance": dict(raw.get("provenance") or {}),
|
|
}
|
|
if item["kind"] == "attachment" and owner not in {"files", "dms"}:
|
|
raise WikiStoreError(
|
|
"Wiki attachments must reference Files or DMS resources."
|
|
)
|
|
links.append(item)
|
|
return links
|
|
|
|
|
|
def _acl(values: object) -> list[str]:
|
|
if not isinstance(values, Sequence) or isinstance(values, (str, bytes)):
|
|
raise WikiStoreError("ACL tokens must be a list.")
|
|
tokens = tuple(dict.fromkeys(_bounded(item, "ACL token", 320) for item in values))
|
|
if len(tokens) > 500:
|
|
raise WikiStoreError("An ACL supports at most 500 subject tokens.")
|
|
for token in tokens:
|
|
prefix, separator, subject_id = token.partition(":")
|
|
if not separator or prefix not in _ACL_PREFIXES or not subject_id:
|
|
raise WikiStoreError(
|
|
"ACL tokens require a supported subject prefix and identifier."
|
|
)
|
|
return list(tokens)
|
|
|
|
|
|
def _principal_tokens(principal: object) -> tuple[str, ...]:
|
|
values: list[str] = []
|
|
for prefix, attribute in (
|
|
("account", "account_id"),
|
|
("identity", "identity_id"),
|
|
("membership", "membership_id"),
|
|
("function_assignment", "acting_assignment_id"),
|
|
):
|
|
value = getattr(principal, attribute, None)
|
|
if str(value or "").strip():
|
|
values.append(f"{prefix}:{value}")
|
|
for prefix, attribute in (
|
|
("group", "group_ids"),
|
|
("role", "role_ids"),
|
|
("function_assignment", "function_assignment_ids"),
|
|
):
|
|
values.extend(
|
|
f"{prefix}:{item}"
|
|
for item in getattr(principal, attribute, ()) or ()
|
|
if str(item or "").strip()
|
|
)
|
|
return tuple(dict.fromkeys(values))
|
|
|
|
|
|
def _search_text(value: Mapping[str, object]) -> str:
|
|
links = value.get("links") or ()
|
|
link_labels = [
|
|
str(item.get("label") or item.get("resource_id") or "")
|
|
for item in links
|
|
if isinstance(item, Mapping)
|
|
]
|
|
return "\n".join(
|
|
[
|
|
str(value.get("title") or ""),
|
|
str(value.get("body") or ""),
|
|
str(value.get("summary") or ""),
|
|
*(str(item) for item in value.get("labels") or ()),
|
|
*link_labels,
|
|
]
|
|
).casefold()
|
|
|
|
|
|
def _emit(
|
|
session: Session,
|
|
registry: object | None,
|
|
tenant_id: str,
|
|
event_type: str,
|
|
resource_type: str,
|
|
resource_id: str,
|
|
label: str,
|
|
revision: int,
|
|
actor: str | None,
|
|
at: datetime,
|
|
) -> None:
|
|
emit_platform_event(
|
|
session,
|
|
PlatformEvent(
|
|
type=event_type,
|
|
module_id="wiki",
|
|
payload={"resource_id": resource_id, "revision": revision},
|
|
occurred_at=at,
|
|
actor=EventActorRef(type="account", id=actor),
|
|
tenant=EventTenantRef(id=tenant_id),
|
|
subject=EventObjectRef(type=resource_type, id=resource_id, label=label),
|
|
resource=EventObjectRef(type=resource_type, id=resource_id, label=label),
|
|
classification="restricted",
|
|
),
|
|
registry=registry,
|
|
)
|
|
|
|
|
|
def _active_modules(registry: object | None) -> frozenset[str]:
|
|
if registry is None:
|
|
return frozenset()
|
|
method = getattr(registry, "active_module_ids", None)
|
|
return (
|
|
frozenset(str(item) for item in method()) if callable(method) else frozenset()
|
|
)
|
|
|
|
|
|
def _tenant(principal: object) -> str:
|
|
value = str(getattr(principal, "tenant_id", "") or "").strip()
|
|
if not value:
|
|
raise WikiStoreError("Wiki operations require a tenant-bound principal.")
|
|
return value
|
|
|
|
|
|
def _actor(principal: object) -> str | None:
|
|
for value in (
|
|
getattr(principal, "account_id", None),
|
|
getattr(principal, "identity_id", None),
|
|
getattr(getattr(principal, "user", None), "id", None),
|
|
):
|
|
if str(value or "").strip():
|
|
return str(value)
|
|
return None
|
|
|
|
|
|
def _has_scope(principal: object, scope: str) -> bool:
|
|
method = getattr(principal, "has", None)
|
|
return (
|
|
bool(method(scope))
|
|
if callable(method)
|
|
else scopes_grant_compatible(
|
|
frozenset(getattr(principal, "scopes", ()) or ()), scope
|
|
)
|
|
)
|
|
|
|
|
|
def _has_any_scope(principal: object, *scopes: str) -> bool:
|
|
return any(_has_scope(principal, scope) for scope in scopes)
|
|
|
|
|
|
def _require_scope(principal: object, scope: str) -> None:
|
|
if not _has_scope(principal, scope):
|
|
raise PermissionError(f"Wiki operation requires {scope}.")
|
|
|
|
|
|
def _require_any_scope(principal: object, *scopes: str) -> None:
|
|
if not any(_has_scope(principal, item) for item in scopes):
|
|
raise PermissionError("Wiki operation is not permitted.")
|
|
|
|
|
|
def _choice(
|
|
value: object, choices: Sequence[str] | set[str] | frozenset[str], label: str
|
|
) -> str:
|
|
clean = str(value or "").strip()
|
|
if clean not in choices:
|
|
raise WikiStoreError(f"Unsupported Wiki {label}.")
|
|
return clean
|
|
|
|
|
|
def _visibility(value: object) -> str:
|
|
return _choice(value, _VISIBILITIES, "visibility")
|
|
|
|
|
|
def _restricted_requires_acl(visibility: object, tokens: object) -> None:
|
|
if visibility == "restricted" and not tokens:
|
|
raise WikiStoreError("Restricted Wiki content requires at least one ACL token.")
|
|
|
|
|
|
def _slug(value: object, label: str, maximum: int) -> str:
|
|
clean = _bounded(value, label, maximum).casefold()
|
|
if not _SLUG.fullmatch(clean):
|
|
raise WikiStoreError(
|
|
f"{label} must use lower-case letters, digits, and single hyphens."
|
|
)
|
|
return clean
|
|
|
|
|
|
def _bounded(value: object, label: str, maximum: int) -> str:
|
|
clean = str(value or "").strip()
|
|
if not clean or len(clean) > maximum:
|
|
raise WikiStoreError(f"{label} must contain 1 to {maximum} characters.")
|
|
return clean
|
|
|
|
|
|
def _optional(value: object, maximum: int) -> str:
|
|
clean = str(value or "").strip()
|
|
if len(clean) > maximum:
|
|
raise WikiStoreError(f"Wiki text exceeds {maximum} characters.")
|
|
return clean
|
|
|
|
|
|
def _aware(value: datetime, label: str) -> datetime:
|
|
if value.tzinfo is None or value.utcoffset() is None:
|
|
raise WikiStoreError(f"{label} must include a timezone.")
|
|
return value
|
|
|
|
|
|
def _sha(value: object) -> str:
|
|
def normalize(item: object) -> Any:
|
|
if isinstance(item, datetime):
|
|
return item.astimezone(UTC).isoformat()
|
|
if isinstance(item, Mapping):
|
|
return {str(key): normalize(val) for key, val in item.items()}
|
|
if isinstance(item, (list, tuple)):
|
|
return [normalize(val) for val in item]
|
|
return item
|
|
|
|
return hashlib.sha256(
|
|
json.dumps(
|
|
normalize(value), sort_keys=True, separators=(",", ":"), ensure_ascii=True
|
|
).encode()
|
|
).hexdigest()
|
|
|
|
|
|
def _iso(value: datetime | None) -> str | None:
|
|
if value is not None and (value.tzinfo is None or value.utcoffset() is None):
|
|
value = value.replace(tzinfo=UTC)
|
|
return value.isoformat() if value else None
|
|
|
|
|
|
__all__ = [
|
|
"ADMIN_SCOPE",
|
|
"CAPABILITY_WIKI_REGISTRY",
|
|
"COMMENT_SCOPE",
|
|
"PUBLISH_SCOPE",
|
|
"READ_SCOPE",
|
|
"SqlWikiRegistry",
|
|
"WRITE_SCOPE",
|
|
"WikiConflictError",
|
|
"WikiNotFoundError",
|
|
"WikiStoreError",
|
|
"add_comment",
|
|
"archive_page",
|
|
"archive_space",
|
|
"can_read_page",
|
|
"compare_revisions",
|
|
"create_page",
|
|
"create_space",
|
|
"get_page",
|
|
"get_space",
|
|
"integration_availability",
|
|
"list_comments",
|
|
"list_pages",
|
|
"list_spaces",
|
|
"page_revisions",
|
|
"publish_page",
|
|
"update_page",
|
|
"update_space",
|
|
]
|