|
|
@@ -0,0 +1,1070 @@
|
|
|
+"""Thin unified work contract over existing governance source state machines."""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import copy
|
|
|
+import re
|
|
|
+import uuid
|
|
|
+from collections.abc import Callable
|
|
|
+from dataclasses import dataclass
|
|
|
+from datetime import datetime, timedelta
|
|
|
+from typing import Any
|
|
|
+
|
|
|
+from app.core.common.identifiers import new_governance_uid
|
|
|
+from app.core.common.timezone_utils import now_china
|
|
|
+
|
|
|
+SUBJECT_TYPES = frozenset(
|
|
|
+ {"quality_issue", "semantic_governance", "data_product", "agent"}
|
|
|
+)
|
|
|
+TASK_TYPES = frozenset(
|
|
|
+ {
|
|
|
+ "approval",
|
|
|
+ "quality_issue",
|
|
|
+ "semantic_governance",
|
|
|
+ "data_product_approval",
|
|
|
+ "agent_approval",
|
|
|
+ "governance_work_order",
|
|
|
+ "release",
|
|
|
+ "high_risk",
|
|
|
+ }
|
|
|
+)
|
|
|
+APPROVAL_MODES = frozenset({"single", "any", "all", "dual_control"})
|
|
|
+PRIORITIES = frozenset({"low", "medium", "high", "critical"})
|
|
|
+DECISIONS = frozenset({"approve", "reject"})
|
|
|
+TIMEOUT_ACTIONS = frozenset({"escalate", "close"})
|
|
|
+NOTIFICATION_CHANNELS = frozenset({"in_app", "email"})
|
|
|
+TEMPLATE_STATUSES = frozenset({"active", "retired"})
|
|
|
+CONDITION_FIELDS = frozenset(
|
|
|
+ {
|
|
|
+ "business_domain_uid",
|
|
|
+ "risk_level",
|
|
|
+ "sensitivity_level",
|
|
|
+ "environment",
|
|
|
+ "amount",
|
|
|
+ }
|
|
|
+)
|
|
|
+CONDITION_OPERATORS = frozenset({"eq", "in", "gte", "lte"})
|
|
|
+ACTIVE_TASK_STATUSES = frozenset(
|
|
|
+ {"pending", "in_progress", "pending_review", "reopened"}
|
|
|
+)
|
|
|
+CODE_PATTERN = re.compile(r"^[A-Z][A-Z0-9_]{2,119}$")
|
|
|
+HASH_PATTERN = re.compile(r"^[0-9a-f]{64}$")
|
|
|
+
|
|
|
+
|
|
|
+@dataclass(frozen=True)
|
|
|
+class DeliveryResult:
|
|
|
+ status: str
|
|
|
+ attempts: int
|
|
|
+ available_at: datetime | None = None
|
|
|
+ last_error: str | None = None
|
|
|
+
|
|
|
+
|
|
|
+def _closed(value: Any, allowed: set[str], label: str) -> dict[str, Any]:
|
|
|
+ if not isinstance(value, dict):
|
|
|
+ raise ValueError(f"{label} must be an object")
|
|
|
+ unknown = sorted(set(value) - allowed)
|
|
|
+ if unknown:
|
|
|
+ raise ValueError(f"{label} contains unsupported fields: {', '.join(unknown)}")
|
|
|
+ return copy.deepcopy(value)
|
|
|
+
|
|
|
+
|
|
|
+def _string(value: Any, label: str, maximum: int = 1000) -> str:
|
|
|
+ if not isinstance(value, str) or not value.strip():
|
|
|
+ raise ValueError(f"{label} is required")
|
|
|
+ result = value.strip()
|
|
|
+ if len(result) > maximum:
|
|
|
+ raise ValueError(f"{label} exceeds {maximum} characters")
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def _optional_string(value: Any, label: str, maximum: int = 1000):
|
|
|
+ if value is None:
|
|
|
+ return None
|
|
|
+ return _string(value, label, maximum)
|
|
|
+
|
|
|
+
|
|
|
+def _uid(value: Any, label: str) -> str:
|
|
|
+ try:
|
|
|
+ return str(uuid.UUID(str(value)))
|
|
|
+ except (TypeError, ValueError, AttributeError) as error:
|
|
|
+ raise ValueError(f"{label} must be a UUID") from error
|
|
|
+
|
|
|
+
|
|
|
+def _time(value: Any, label: str) -> datetime:
|
|
|
+ if isinstance(value, datetime):
|
|
|
+ result = value
|
|
|
+ else:
|
|
|
+ try:
|
|
|
+ result = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
|
|
+ except (TypeError, ValueError) as error:
|
|
|
+ raise ValueError(f"{label} must be an ISO-8601 datetime") from error
|
|
|
+ if result.tzinfo is None:
|
|
|
+ raise ValueError(f"{label} must include a timezone")
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def _list(value: Any, label: str, *, minimum: int = 0) -> list[Any]:
|
|
|
+ if not isinstance(value, list) or len(value) < minimum:
|
|
|
+ raise ValueError(f"{label} must contain at least {minimum} items")
|
|
|
+ return copy.deepcopy(value)
|
|
|
+
|
|
|
+
|
|
|
+def _normalize_condition(value: Any) -> dict[str, Any]:
|
|
|
+ item = _closed(value, {"field", "operator", "value"}, "route condition")
|
|
|
+ field = _string(item.get("field"), "condition field", 50)
|
|
|
+ operator = _string(item.get("operator"), "condition operator", 10)
|
|
|
+ if field not in CONDITION_FIELDS or operator not in CONDITION_OPERATORS:
|
|
|
+ raise ValueError("unsupported route condition")
|
|
|
+ condition_value = item.get("value")
|
|
|
+ if operator == "in" and not isinstance(condition_value, list):
|
|
|
+ raise ValueError("in condition requires an array value")
|
|
|
+ if operator in {"gte", "lte"}:
|
|
|
+ try:
|
|
|
+ condition_value = float(condition_value)
|
|
|
+ except (TypeError, ValueError) as error:
|
|
|
+ raise ValueError("numeric route condition is invalid") from error
|
|
|
+ return {"field": field, "operator": operator, "value": condition_value}
|
|
|
+
|
|
|
+
|
|
|
+def _normalize_route(value: Any, *, priority: int = 9999) -> dict[str, Any]:
|
|
|
+ route = _closed(
|
|
|
+ value,
|
|
|
+ {
|
|
|
+ "priority",
|
|
|
+ "conditions",
|
|
|
+ "approval_mode",
|
|
|
+ "reviewer_uids",
|
|
|
+ "min_approvals",
|
|
|
+ "due_hours",
|
|
|
+ "timeout_action",
|
|
|
+ "notification_channels",
|
|
|
+ },
|
|
|
+ "workflow route",
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ normalized_priority = int(route.get("priority", priority))
|
|
|
+ minimum = int(route.get("min_approvals", 1))
|
|
|
+ due_hours = int(route.get("due_hours", 24))
|
|
|
+ except (TypeError, ValueError) as error:
|
|
|
+ raise ValueError("workflow route numeric field is invalid") from error
|
|
|
+ if normalized_priority < 0 or due_hours < 1 or due_hours > 8760:
|
|
|
+ raise ValueError("workflow route priority or due time is invalid")
|
|
|
+ approval_mode = _string(route.get("approval_mode"), "approval_mode", 30)
|
|
|
+ if approval_mode not in APPROVAL_MODES:
|
|
|
+ raise ValueError("unsupported approval mode")
|
|
|
+ reviewers = sorted(
|
|
|
+ {
|
|
|
+ _uid(item, "reviewer uid")
|
|
|
+ for item in _list(route.get("reviewer_uids"), "reviewer_uids", minimum=1)
|
|
|
+ }
|
|
|
+ )
|
|
|
+ if minimum < 1 or minimum > len(reviewers):
|
|
|
+ raise ValueError("min approvals is outside reviewer range")
|
|
|
+ if approval_mode == "single" and (len(reviewers) != 1 or minimum != 1):
|
|
|
+ raise ValueError("single approval requires exactly one reviewer")
|
|
|
+ if approval_mode == "all" and minimum != len(reviewers):
|
|
|
+ raise ValueError("all approval requires every reviewer")
|
|
|
+ if approval_mode == "dual_control" and (len(reviewers) < 2 or minimum < 2):
|
|
|
+ raise ValueError("dual control requires two independent reviewers")
|
|
|
+ timeout_action = _string(
|
|
|
+ route.get("timeout_action", "escalate"), "timeout_action", 20
|
|
|
+ )
|
|
|
+ if timeout_action not in TIMEOUT_ACTIONS:
|
|
|
+ raise ValueError("unsupported timeout action")
|
|
|
+ channels = sorted(
|
|
|
+ {
|
|
|
+ _string(item, "notification channel", 20)
|
|
|
+ for item in route.get("notification_channels", ["in_app"])
|
|
|
+ }
|
|
|
+ )
|
|
|
+ if not channels or not set(channels) <= NOTIFICATION_CHANNELS:
|
|
|
+ raise ValueError("unsupported notification channel")
|
|
|
+ return {
|
|
|
+ "priority": normalized_priority,
|
|
|
+ "conditions": [
|
|
|
+ _normalize_condition(item) for item in route.get("conditions", [])
|
|
|
+ ],
|
|
|
+ "approval_mode": approval_mode,
|
|
|
+ "reviewer_uids": reviewers,
|
|
|
+ "min_approvals": minimum,
|
|
|
+ "due_hours": due_hours,
|
|
|
+ "timeout_action": timeout_action,
|
|
|
+ "notification_channels": channels,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def _matches(route: dict[str, Any], context: dict[str, Any]) -> bool:
|
|
|
+ for condition in route["conditions"]:
|
|
|
+ actual = context.get(condition["field"])
|
|
|
+ expected = condition["value"]
|
|
|
+ operator = condition["operator"]
|
|
|
+ if operator == "eq" and actual != expected:
|
|
|
+ return False
|
|
|
+ if operator == "in" and actual not in expected:
|
|
|
+ return False
|
|
|
+ if operator in {"gte", "lte"}:
|
|
|
+ try:
|
|
|
+ numeric = float(actual)
|
|
|
+ except (TypeError, ValueError):
|
|
|
+ return False
|
|
|
+ if operator == "gte" and numeric < expected:
|
|
|
+ return False
|
|
|
+ if operator == "lte" and numeric > expected:
|
|
|
+ return False
|
|
|
+ return True
|
|
|
+
|
|
|
+
|
|
|
+def _redact_error(error: Exception) -> str:
|
|
|
+ message = re.sub(
|
|
|
+ r"(?i)(secret[-_ ]?token|api[-_ ]?key|authorization|bearer)(?:[=: ]+\S+)?",
|
|
|
+ "[redacted]",
|
|
|
+ str(error),
|
|
|
+ )
|
|
|
+ return message[:1000]
|
|
|
+
|
|
|
+
|
|
|
+class UnifiedWorkCenterService:
|
|
|
+ """Coordinate work evidence while source modules remain authoritative."""
|
|
|
+
|
|
|
+ def __init__(
|
|
|
+ self,
|
|
|
+ repository,
|
|
|
+ *,
|
|
|
+ uid_factory: Callable[[], str] = new_governance_uid,
|
|
|
+ now_factory: Callable[[], datetime] = now_china,
|
|
|
+ commit: Callable[[], None] = lambda: None,
|
|
|
+ rollback: Callable[[], None] = lambda: None,
|
|
|
+ ):
|
|
|
+ self.repository = repository
|
|
|
+ self.uid_factory = uid_factory
|
|
|
+ self.now_factory = now_factory
|
|
|
+ self.commit = commit
|
|
|
+ self.rollback = rollback
|
|
|
+
|
|
|
+ def create_workflow(self, payload: Any, *, actor_uid: str) -> dict[str, Any]:
|
|
|
+ body = _closed(
|
|
|
+ payload,
|
|
|
+ {"code", "name", "subject_types", "routes", "default_route"},
|
|
|
+ "workflow definition",
|
|
|
+ )
|
|
|
+ code = _string(body.get("code"), "code", 120).upper()
|
|
|
+ if not CODE_PATTERN.fullmatch(code):
|
|
|
+ raise ValueError("workflow code is invalid")
|
|
|
+ subject_types = sorted(
|
|
|
+ {
|
|
|
+ _string(item, "subject type", 40)
|
|
|
+ for item in _list(
|
|
|
+ body.get("subject_types"), "subject_types", minimum=1
|
|
|
+ )
|
|
|
+ }
|
|
|
+ )
|
|
|
+ if not set(subject_types) <= SUBJECT_TYPES:
|
|
|
+ raise ValueError("unsupported workflow subject type")
|
|
|
+ routes = sorted(
|
|
|
+ [_normalize_route(item) for item in body.get("routes", [])],
|
|
|
+ key=lambda item: item["priority"],
|
|
|
+ )
|
|
|
+ default_route = _normalize_route(body.get("default_route"), priority=9999)
|
|
|
+ users = {
|
|
|
+ user
|
|
|
+ for route in [*routes, default_route]
|
|
|
+ for user in route["reviewer_uids"]
|
|
|
+ }
|
|
|
+ if self.repository.users_available(users) != users:
|
|
|
+ raise ValueError("workflow reviewer is unknown or disabled")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ workflow_uid = self.uid_factory()
|
|
|
+ version_uid = self.uid_factory()
|
|
|
+ workflow = {
|
|
|
+ "uid": workflow_uid,
|
|
|
+ "code": code,
|
|
|
+ "name": _string(body.get("name"), "name", 300),
|
|
|
+ "status": "draft",
|
|
|
+ "current_version": 1,
|
|
|
+ "active_version_uid": None,
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now,
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ version = {
|
|
|
+ "uid": version_uid,
|
|
|
+ "workflow_uid": workflow_uid,
|
|
|
+ "version": 1,
|
|
|
+ "status": "draft",
|
|
|
+ "definition": {
|
|
|
+ "subject_types": subject_types,
|
|
|
+ "routes": routes,
|
|
|
+ "default_route": default_route,
|
|
|
+ },
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now,
|
|
|
+ "published_by": None,
|
|
|
+ "published_at": None,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ self.repository.create_workflow(workflow, version)
|
|
|
+ self.commit()
|
|
|
+ return {**workflow, "latest_version": version}
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def publish_workflow(
|
|
|
+ self, workflow_uid: str, *, expected_version: int, actor_uid: str
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ uid = _uid(workflow_uid, "workflow_uid")
|
|
|
+ workflow = self.repository.get_workflow(uid)
|
|
|
+ if workflow is None:
|
|
|
+ raise LookupError("workflow was not found")
|
|
|
+ if int(workflow["current_version"]) != int(expected_version):
|
|
|
+ raise RuntimeError("workflow version conflict")
|
|
|
+ version = self.repository.workflow_version(uid, int(expected_version))
|
|
|
+ if version is None or version["status"] != "draft":
|
|
|
+ raise RuntimeError("workflow version is not publishable")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ published_version = {
|
|
|
+ **version,
|
|
|
+ "status": "published",
|
|
|
+ "published_by": actor,
|
|
|
+ "published_at": now,
|
|
|
+ }
|
|
|
+ published = {
|
|
|
+ **workflow,
|
|
|
+ "status": "published",
|
|
|
+ "active_version_uid": version["uid"],
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ self.repository.publish_workflow(
|
|
|
+ published, published_version, int(expected_version)
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return {**published, "active_version": published_version}
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def revise_workflow(
|
|
|
+ self,
|
|
|
+ workflow_uid: str,
|
|
|
+ payload: Any,
|
|
|
+ *,
|
|
|
+ expected_version: int,
|
|
|
+ actor_uid: str,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ body = _closed(
|
|
|
+ payload,
|
|
|
+ {"subject_types", "routes", "default_route"},
|
|
|
+ "workflow revision",
|
|
|
+ )
|
|
|
+ uid = _uid(workflow_uid, "workflow_uid")
|
|
|
+ workflow = self.repository.get_workflow(uid)
|
|
|
+ if workflow is None:
|
|
|
+ raise LookupError("workflow was not found")
|
|
|
+ if int(workflow["current_version"]) != int(expected_version):
|
|
|
+ raise RuntimeError("workflow version conflict")
|
|
|
+ subject_types = sorted(
|
|
|
+ {
|
|
|
+ _string(item, "subject type", 40)
|
|
|
+ for item in _list(body.get("subject_types"), "subject_types", minimum=1)
|
|
|
+ }
|
|
|
+ )
|
|
|
+ if not set(subject_types) <= SUBJECT_TYPES:
|
|
|
+ raise ValueError("unsupported workflow subject type")
|
|
|
+ routes = sorted(
|
|
|
+ [_normalize_route(item) for item in body.get("routes", [])],
|
|
|
+ key=lambda item: item["priority"],
|
|
|
+ )
|
|
|
+ default_route = _normalize_route(body.get("default_route"), priority=9999)
|
|
|
+ reviewers = {
|
|
|
+ reviewer
|
|
|
+ for route in [*routes, default_route]
|
|
|
+ for reviewer in route["reviewer_uids"]
|
|
|
+ }
|
|
|
+ if self.repository.users_available(reviewers) != reviewers:
|
|
|
+ raise ValueError("workflow reviewer is unknown or disabled")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ next_version = int(expected_version) + 1
|
|
|
+ revised = {
|
|
|
+ **workflow,
|
|
|
+ "status": "published" if workflow.get("active_version_uid") else "draft",
|
|
|
+ "current_version": next_version,
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ version = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "workflow_uid": uid,
|
|
|
+ "version": next_version,
|
|
|
+ "status": "draft",
|
|
|
+ "definition": {
|
|
|
+ "subject_types": subject_types,
|
|
|
+ "routes": routes,
|
|
|
+ "default_route": default_route,
|
|
|
+ },
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now,
|
|
|
+ "published_by": None,
|
|
|
+ "published_at": None,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ self.repository.revise_workflow(revised, version, int(expected_version))
|
|
|
+ self.commit()
|
|
|
+ return {**revised, "latest_version": version}
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def list_workflows(self) -> list[dict[str, Any]]:
|
|
|
+ return self.repository.list_workflows()
|
|
|
+
|
|
|
+ def _published_definition(self, workflow_uid: str):
|
|
|
+ workflow = self.repository.get_workflow(_uid(workflow_uid, "workflow_uid"))
|
|
|
+ if workflow is None or workflow["status"] != "published":
|
|
|
+ raise LookupError("published workflow was not found")
|
|
|
+ if workflow.get("active_version_uid") and hasattr(
|
|
|
+ self.repository, "workflow_version_by_uid"
|
|
|
+ ):
|
|
|
+ version = self.repository.workflow_version_by_uid(
|
|
|
+ workflow["active_version_uid"]
|
|
|
+ )
|
|
|
+ else:
|
|
|
+ version = self.repository.workflow_version(
|
|
|
+ workflow["uid"], int(workflow["current_version"])
|
|
|
+ )
|
|
|
+ if version is None or version["status"] != "published":
|
|
|
+ raise RuntimeError("active workflow version is unavailable")
|
|
|
+ return workflow, version
|
|
|
+
|
|
|
+ def _select_route(self, definition, context):
|
|
|
+ return next(
|
|
|
+ (
|
|
|
+ copy.deepcopy(route)
|
|
|
+ for route in definition["routes"]
|
|
|
+ if _matches(route, context)
|
|
|
+ ),
|
|
|
+ copy.deepcopy(definition["default_route"]),
|
|
|
+ )
|
|
|
+
|
|
|
+ def create_task(self, payload: Any, *, actor_uid: str) -> dict[str, Any]:
|
|
|
+ body = _closed(
|
|
|
+ payload,
|
|
|
+ {
|
|
|
+ "workflow_uid",
|
|
|
+ "task_type",
|
|
|
+ "subject_type",
|
|
|
+ "subject_uid",
|
|
|
+ "source_type",
|
|
|
+ "source_uid",
|
|
|
+ "title",
|
|
|
+ "description",
|
|
|
+ "priority",
|
|
|
+ "business_domain_uid",
|
|
|
+ "context",
|
|
|
+ },
|
|
|
+ "unified task",
|
|
|
+ )
|
|
|
+ workflow, version = self._published_definition(body.get("workflow_uid"))
|
|
|
+ task_type = _string(body.get("task_type"), "task_type", 40)
|
|
|
+ subject_type = _string(body.get("subject_type"), "subject_type", 40)
|
|
|
+ if task_type not in TASK_TYPES or subject_type not in SUBJECT_TYPES:
|
|
|
+ raise ValueError("unsupported task or subject type")
|
|
|
+ if subject_type not in version["definition"]["subject_types"]:
|
|
|
+ raise ValueError("workflow does not support the subject type")
|
|
|
+ context = body.get("context") or {}
|
|
|
+ if not isinstance(context, dict):
|
|
|
+ raise ValueError("task context must be an object")
|
|
|
+ route = self._select_route(version["definition"], context)
|
|
|
+ if self.repository.users_available(route["reviewer_uids"]) != set(
|
|
|
+ route["reviewer_uids"]
|
|
|
+ ):
|
|
|
+ raise ValueError("task reviewer is unknown or disabled")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ now = self.now_factory()
|
|
|
+ task_uid = self.uid_factory()
|
|
|
+ task = {
|
|
|
+ "uid": task_uid,
|
|
|
+ "task_code": f"GWT-{now:%Y%m%d}-{task_uid[-8:].upper()}",
|
|
|
+ "workflow_uid": workflow["uid"],
|
|
|
+ "workflow_version": int(version["version"]),
|
|
|
+ "task_type": task_type,
|
|
|
+ "subject_type": subject_type,
|
|
|
+ "subject_uid": _string(body.get("subject_uid"), "subject_uid", 200),
|
|
|
+ "source_type": _string(body.get("source_type"), "source_type", 80),
|
|
|
+ "source_uid": _string(body.get("source_uid"), "source_uid", 200),
|
|
|
+ "title": _string(body.get("title"), "title", 300),
|
|
|
+ "description": _string(body.get("description"), "description", 2000),
|
|
|
+ "business_domain_uid": _optional_string(
|
|
|
+ body.get("business_domain_uid"), "business_domain_uid", 200
|
|
|
+ ),
|
|
|
+ "priority": _string(body.get("priority"), "priority", 20),
|
|
|
+ "status": "pending",
|
|
|
+ "assignee_uid": route["reviewer_uids"][0],
|
|
|
+ "due_at": (now + timedelta(hours=route["due_hours"])).isoformat(),
|
|
|
+ "escalation_level": 0,
|
|
|
+ "current_version": 1,
|
|
|
+ "route_snapshot": route,
|
|
|
+ "context": copy.deepcopy(context),
|
|
|
+ "source_state_unchanged": True,
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now.isoformat(),
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": now.isoformat(),
|
|
|
+ "closed_at": None,
|
|
|
+ }
|
|
|
+ if task["priority"] not in PRIORITIES:
|
|
|
+ raise ValueError("unsupported task priority")
|
|
|
+ participants = [
|
|
|
+ {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "task_uid": task_uid,
|
|
|
+ "user_uid": reviewer,
|
|
|
+ "sequence": index + 1,
|
|
|
+ "status": "pending",
|
|
|
+ "transferred_from_uid": None,
|
|
|
+ "created_at": now.isoformat(),
|
|
|
+ }
|
|
|
+ for index, reviewer in enumerate(route["reviewer_uids"])
|
|
|
+ ]
|
|
|
+ try:
|
|
|
+ saved = self.repository.create_task(task, participants)
|
|
|
+ created = saved.pop("_created", True)
|
|
|
+ if created:
|
|
|
+ self._notify_task(saved, participants, event_type="task_created")
|
|
|
+ self.commit()
|
|
|
+ return saved
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def _notify_task(self, task, participants, *, event_type):
|
|
|
+ now = self.now_factory()
|
|
|
+ for recipient in sorted({item["user_uid"] for item in participants}):
|
|
|
+ for channel in task["route_snapshot"]["notification_channels"]:
|
|
|
+ notification_uid = self.uid_factory()
|
|
|
+ record = {
|
|
|
+ "uid": notification_uid,
|
|
|
+ "event_key": f"{event_type}:{task['uid']}:{recipient}:{channel}",
|
|
|
+ "event_type": event_type,
|
|
|
+ "recipient_uid": recipient,
|
|
|
+ "channel": channel,
|
|
|
+ "subject": task["title"],
|
|
|
+ "body": task["description"],
|
|
|
+ "related_task_uid": task["uid"],
|
|
|
+ "status": "delivered" if channel == "in_app" else "pending",
|
|
|
+ "attempts": 0,
|
|
|
+ "max_attempts": 5,
|
|
|
+ "next_attempt_at": now.isoformat(),
|
|
|
+ "last_error": None,
|
|
|
+ "delivered_at": now.isoformat() if channel == "in_app" else None,
|
|
|
+ "read_at": None,
|
|
|
+ "created_at": now.isoformat(),
|
|
|
+ }
|
|
|
+ if hasattr(self.repository, "prepare_notification"):
|
|
|
+ record = self.repository.prepare_notification(record, task)
|
|
|
+ self.repository.create_notification(record)
|
|
|
+
|
|
|
+ def list_tasks(self, **filters) -> list[dict[str, Any]]:
|
|
|
+ return self.repository.list_tasks(**filters)
|
|
|
+
|
|
|
+ def task_detail(
|
|
|
+ self,
|
|
|
+ uid: str,
|
|
|
+ *,
|
|
|
+ requester_uid: str | None = None,
|
|
|
+ can_manage: bool = False,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ result = self.repository.task_detail(_uid(uid, "task_uid"))
|
|
|
+ if result is None:
|
|
|
+ raise LookupError("task was not found")
|
|
|
+ if requester_uid and not can_manage:
|
|
|
+ actor = _uid(requester_uid, "requester_uid")
|
|
|
+ visible = {result["created_by"], result["assignee_uid"]} | {
|
|
|
+ item["user_uid"] for item in result.get("participants", [])
|
|
|
+ }
|
|
|
+ if actor not in visible:
|
|
|
+ raise LookupError("task was not found")
|
|
|
+ return result
|
|
|
+
|
|
|
+ def _require_task_collaborator(self, task_uid: str, actor_uid: str):
|
|
|
+ task = self.repository.get_task(task_uid)
|
|
|
+ if task is None:
|
|
|
+ raise LookupError("task was not found")
|
|
|
+ participant_uids = {
|
|
|
+ item["user_uid"] for item in self.repository.participants_for_task(task_uid)
|
|
|
+ }
|
|
|
+ if actor_uid not in {
|
|
|
+ task["created_by"],
|
|
|
+ task["assignee_uid"],
|
|
|
+ *participant_uids,
|
|
|
+ }:
|
|
|
+ raise PermissionError("actor is not a task collaborator")
|
|
|
+ return task
|
|
|
+
|
|
|
+ def review_task(
|
|
|
+ self,
|
|
|
+ task_uid: str,
|
|
|
+ payload: Any,
|
|
|
+ *,
|
|
|
+ expected_version: int,
|
|
|
+ actor_uid: str,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ body = _closed(payload, {"decision", "reason"}, "task review")
|
|
|
+ uid = _uid(task_uid, "task_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ task = self.repository.get_task(uid)
|
|
|
+ if task is None:
|
|
|
+ raise LookupError("task was not found")
|
|
|
+ if task["status"] not in ACTIVE_TASK_STATUSES:
|
|
|
+ raise RuntimeError("task is not reviewable")
|
|
|
+ participants = self.repository.participants_for_task(uid)
|
|
|
+ if not any(
|
|
|
+ item["user_uid"] == actor and item["status"] == "pending"
|
|
|
+ for item in participants
|
|
|
+ ):
|
|
|
+ raise PermissionError("reviewer is not a pending participant")
|
|
|
+ mode = task["route_snapshot"]["approval_mode"]
|
|
|
+ if mode == "dual_control" and actor == task["created_by"]:
|
|
|
+ raise PermissionError("dual control creator cannot review")
|
|
|
+ decision = _string(body.get("decision"), "decision", 20)
|
|
|
+ if decision not in DECISIONS:
|
|
|
+ raise ValueError("unsupported review decision")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ self.repository.add_review(
|
|
|
+ {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "task_uid": uid,
|
|
|
+ "reviewer_uid": actor,
|
|
|
+ "decision": decision,
|
|
|
+ "reason": _string(body.get("reason"), "reason", 1000),
|
|
|
+ "created_at": now,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ reviews = self.repository.reviews_for_task(uid)
|
|
|
+ approvals = {item["reviewer_uid"] for item in reviews if item["decision"] == "approve"}
|
|
|
+ if any(item["decision"] == "reject" for item in reviews):
|
|
|
+ status = "rejected"
|
|
|
+ elif mode == "all":
|
|
|
+ status = "approved" if len(approvals) == len(participants) else "pending"
|
|
|
+ else:
|
|
|
+ status = (
|
|
|
+ "approved"
|
|
|
+ if len(approvals) >= int(task["route_snapshot"]["min_approvals"])
|
|
|
+ else "pending"
|
|
|
+ )
|
|
|
+ task.update({"status": status, "updated_by": actor, "updated_at": now})
|
|
|
+ try:
|
|
|
+ result = self.repository.update_task(
|
|
|
+ task,
|
|
|
+ int(expected_version),
|
|
|
+ "reviewed",
|
|
|
+ actor,
|
|
|
+ {"decision": decision, "resulting_status": status},
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def transfer_review(
|
|
|
+ self,
|
|
|
+ task_uid: str,
|
|
|
+ payload: Any,
|
|
|
+ *,
|
|
|
+ expected_version: int,
|
|
|
+ actor_uid: str,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ body = _closed(
|
|
|
+ payload,
|
|
|
+ {"source_user_uid", "target_user_uid", "reason"},
|
|
|
+ "review transfer",
|
|
|
+ )
|
|
|
+ uid = _uid(task_uid, "task_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ source = _uid(body.get("source_user_uid"), "source_user_uid")
|
|
|
+ target = _uid(body.get("target_user_uid"), "target_user_uid")
|
|
|
+ if actor != source:
|
|
|
+ raise PermissionError("only the pending reviewer can transfer")
|
|
|
+ if source == target or self.repository.users_available([target]) != {target}:
|
|
|
+ raise ValueError("transfer target is invalid or unavailable")
|
|
|
+ task = self.repository.get_task(uid)
|
|
|
+ if task is None:
|
|
|
+ raise LookupError("task was not found")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ try:
|
|
|
+ self.repository.replace_participant(uid, source, target, actor)
|
|
|
+ task.update(
|
|
|
+ {"assignee_uid": target, "updated_by": actor, "updated_at": now}
|
|
|
+ )
|
|
|
+ result = self.repository.update_task(
|
|
|
+ task,
|
|
|
+ int(expected_version),
|
|
|
+ "review_transferred",
|
|
|
+ actor,
|
|
|
+ {"source_user_uid": source, "target_user_uid": target, "reason": _string(body.get("reason"), "reason", 500)},
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def add_comment(self, task_uid: str, payload: Any, *, actor_uid: str):
|
|
|
+ body = _closed(payload, {"content", "mentions"}, "task comment")
|
|
|
+ uid = _uid(task_uid, "task_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ self._require_task_collaborator(uid, actor)
|
|
|
+ mentions = sorted(
|
|
|
+ {_uid(item, "mention uid") for item in body.get("mentions", [])}
|
|
|
+ )
|
|
|
+ if self.repository.users_available(mentions) != set(mentions):
|
|
|
+ raise ValueError("mentioned user is unavailable")
|
|
|
+ record = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "task_uid": uid,
|
|
|
+ "content": _string(body.get("content"), "content", 4000),
|
|
|
+ "mentions": mentions,
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": self.now_factory().isoformat(),
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ result = self.repository.add_comment(record)
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def add_attachment(self, task_uid: str, payload: Any, *, actor_uid: str):
|
|
|
+ body = _closed(
|
|
|
+ payload,
|
|
|
+ {"name", "storage_ref", "content_hash"},
|
|
|
+ "task attachment",
|
|
|
+ )
|
|
|
+ uid = _uid(task_uid, "task_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ self._require_task_collaborator(uid, actor)
|
|
|
+ content_hash = _string(body.get("content_hash"), "content_hash", 64).lower()
|
|
|
+ if not HASH_PATTERN.fullmatch(content_hash):
|
|
|
+ raise ValueError("attachment content hash is invalid")
|
|
|
+ record = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "task_uid": uid,
|
|
|
+ "name": _string(body.get("name"), "name", 300),
|
|
|
+ "storage_ref": _string(body.get("storage_ref"), "storage_ref", 1000),
|
|
|
+ "content_hash": content_hash,
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": self.now_factory().isoformat(),
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ result = self.repository.add_attachment(record)
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def _transition_task(
|
|
|
+ self,
|
|
|
+ task_uid,
|
|
|
+ payload,
|
|
|
+ *,
|
|
|
+ expected_version,
|
|
|
+ actor_uid,
|
|
|
+ action,
|
|
|
+ ):
|
|
|
+ uid = _uid(task_uid, "task_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ task = self.repository.get_task(uid)
|
|
|
+ if task is None:
|
|
|
+ raise LookupError("task was not found")
|
|
|
+ participant_uids = {
|
|
|
+ item["user_uid"] for item in self.repository.participants_for_task(uid)
|
|
|
+ }
|
|
|
+ if actor not in {task["created_by"], task["assignee_uid"], *participant_uids}:
|
|
|
+ raise PermissionError("actor is not a task collaborator")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ if action == "closed":
|
|
|
+ body = _closed(payload, {"resolution", "evidence_refs"}, "task closure")
|
|
|
+ evidence = _list(body.get("evidence_refs"), "evidence_refs", minimum=1)
|
|
|
+ task.update({"status": "closed", "closed_at": now})
|
|
|
+ audit_payload = {
|
|
|
+ "resolution": _string(body.get("resolution"), "resolution", 2000),
|
|
|
+ "evidence_refs": [_string(item, "evidence ref", 1000) for item in evidence],
|
|
|
+ }
|
|
|
+ else:
|
|
|
+ if task["status"] != "closed":
|
|
|
+ raise RuntimeError("only a closed task can be reopened")
|
|
|
+ body = _closed(payload, {"reason"}, "task reopening")
|
|
|
+ task.update({"status": "reopened", "closed_at": None})
|
|
|
+ audit_payload = {"reason": _string(body.get("reason"), "reason", 1000)}
|
|
|
+ task.update({"updated_by": actor, "updated_at": now})
|
|
|
+ try:
|
|
|
+ result = self.repository.update_task(
|
|
|
+ task,
|
|
|
+ int(expected_version),
|
|
|
+ action,
|
|
|
+ actor,
|
|
|
+ audit_payload,
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def close_task(self, task_uid, payload, *, expected_version, actor_uid):
|
|
|
+ return self._transition_task(
|
|
|
+ task_uid,
|
|
|
+ payload,
|
|
|
+ expected_version=expected_version,
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ action="closed",
|
|
|
+ )
|
|
|
+
|
|
|
+ def reopen_task(self, task_uid, payload, *, expected_version, actor_uid):
|
|
|
+ return self._transition_task(
|
|
|
+ task_uid,
|
|
|
+ payload,
|
|
|
+ expected_version=expected_version,
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ action="reopened",
|
|
|
+ )
|
|
|
+
|
|
|
+ def process_timeouts(
|
|
|
+ self, *, at: datetime | str | None = None, actor_uid: str
|
|
|
+ ) -> list[dict[str, Any]]:
|
|
|
+ timestamp = _time(at, "at") if at is not None else self.now_factory()
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ results = []
|
|
|
+ try:
|
|
|
+ for task in self.repository.overdue_tasks(timestamp):
|
|
|
+ action = task["route_snapshot"]["timeout_action"]
|
|
|
+ if action == "close":
|
|
|
+ task.update({"status": "closed", "closed_at": timestamp.isoformat()})
|
|
|
+ event = "timeout_closed"
|
|
|
+ else:
|
|
|
+ task["escalation_level"] = int(task["escalation_level"]) + 1
|
|
|
+ task["priority"] = "critical"
|
|
|
+ task["due_at"] = (
|
|
|
+ timestamp
|
|
|
+ + timedelta(hours=int(task["route_snapshot"]["due_hours"]))
|
|
|
+ ).isoformat()
|
|
|
+ event = "timeout_escalated"
|
|
|
+ task.update(
|
|
|
+ {
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": timestamp.isoformat(),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ results.append(
|
|
|
+ self.repository.update_task(
|
|
|
+ task,
|
|
|
+ int(task["current_version"]),
|
|
|
+ event,
|
|
|
+ actor,
|
|
|
+ {"processed_at": timestamp.isoformat()},
|
|
|
+ )
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return results
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def list_notifications(self, owner_uid: str, *, unread_only: bool = False):
|
|
|
+ return self.repository.list_notifications(
|
|
|
+ _uid(owner_uid, "owner_uid"), unread_only=bool(unread_only)
|
|
|
+ )
|
|
|
+
|
|
|
+ def deliver_notifications(
|
|
|
+ self,
|
|
|
+ channel: str,
|
|
|
+ adapter: Callable[[dict[str, Any]], None],
|
|
|
+ *,
|
|
|
+ at: datetime | str | None = None,
|
|
|
+ limit: int = 50,
|
|
|
+ ) -> list[dict[str, Any]]:
|
|
|
+ normalized_channel = _string(channel, "channel", 20)
|
|
|
+ if normalized_channel != "email":
|
|
|
+ raise ValueError("only email requires external delivery")
|
|
|
+ timestamp = _time(at, "at") if at is not None else self.now_factory()
|
|
|
+ records = self.repository.claim_notifications(
|
|
|
+ normalized_channel, timestamp, max(1, min(int(limit), 200))
|
|
|
+ )
|
|
|
+ results = []
|
|
|
+ try:
|
|
|
+ for notification in records:
|
|
|
+ attempts = int(notification["attempts"]) + 1
|
|
|
+ try:
|
|
|
+ adapter(notification)
|
|
|
+ result = DeliveryResult(status="delivered", attempts=attempts)
|
|
|
+ except Exception as error:
|
|
|
+ if attempts >= int(notification["max_attempts"]):
|
|
|
+ result = DeliveryResult(
|
|
|
+ status="dead_letter",
|
|
|
+ attempts=attempts,
|
|
|
+ last_error=_redact_error(error),
|
|
|
+ )
|
|
|
+ else:
|
|
|
+ result = DeliveryResult(
|
|
|
+ status="pending",
|
|
|
+ attempts=attempts,
|
|
|
+ available_at=timestamp
|
|
|
+ + timedelta(seconds=min(300, 2**attempts)),
|
|
|
+ last_error=_redact_error(error),
|
|
|
+ )
|
|
|
+ results.append(
|
|
|
+ self.repository.record_delivery(notification, result, timestamp)
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return results
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def mark_notification_read(self, uid: str, *, actor_uid: str):
|
|
|
+ try:
|
|
|
+ result = self.repository.mark_notification_read(
|
|
|
+ _uid(uid, "notification_uid"),
|
|
|
+ _uid(actor_uid, "actor_uid"),
|
|
|
+ self.now_factory(),
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def list_notification_templates(self):
|
|
|
+ return self.repository.list_notification_templates()
|
|
|
+
|
|
|
+ def create_notification_template(self, payload: Any, *, actor_uid: str):
|
|
|
+ body = _closed(
|
|
|
+ payload,
|
|
|
+ {"code", "channel", "subject_template", "body_template"},
|
|
|
+ "notification template",
|
|
|
+ )
|
|
|
+ code = _string(body.get("code"), "code", 120).lower()
|
|
|
+ if not re.fullmatch(r"[a-z][a-z0-9_.-]{2,119}", code):
|
|
|
+ raise ValueError("notification template code is invalid")
|
|
|
+ channel = _string(body.get("channel"), "channel", 20)
|
|
|
+ if channel not in NOTIFICATION_CHANNELS:
|
|
|
+ raise ValueError("unsupported notification channel")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ record = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "code": code,
|
|
|
+ "channel": channel,
|
|
|
+ "subject_template": _string(
|
|
|
+ body.get("subject_template"), "subject_template", 500
|
|
|
+ ),
|
|
|
+ "body_template": _string(
|
|
|
+ body.get("body_template"), "body_template", 4000
|
|
|
+ ),
|
|
|
+ "status": "active",
|
|
|
+ "current_version": 1,
|
|
|
+ "created_by": actor,
|
|
|
+ "updated_by": actor,
|
|
|
+ "created_at": now,
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ result = self.repository.create_notification_template(record)
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def revise_notification_template(
|
|
|
+ self,
|
|
|
+ uid: str,
|
|
|
+ payload: Any,
|
|
|
+ *,
|
|
|
+ expected_version: int,
|
|
|
+ actor_uid: str,
|
|
|
+ ):
|
|
|
+ body = _closed(
|
|
|
+ payload,
|
|
|
+ {"subject_template", "body_template", "status"},
|
|
|
+ "notification template revision",
|
|
|
+ )
|
|
|
+ status = _string(body.get("status"), "status", 20)
|
|
|
+ if status not in TEMPLATE_STATUSES:
|
|
|
+ raise ValueError("unsupported notification template status")
|
|
|
+ record = {
|
|
|
+ "uid": _uid(uid, "template_uid"),
|
|
|
+ "subject_template": _string(
|
|
|
+ body.get("subject_template"), "subject_template", 500
|
|
|
+ ),
|
|
|
+ "body_template": _string(
|
|
|
+ body.get("body_template"), "body_template", 4000
|
|
|
+ ),
|
|
|
+ "status": status,
|
|
|
+ "updated_by": _uid(actor_uid, "actor_uid"),
|
|
|
+ "updated_at": self.now_factory().isoformat(),
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ result = self.repository.revise_notification_template(
|
|
|
+ record, int(expected_version)
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def get_notification_preferences(self, user_uid: str):
|
|
|
+ return self.repository.get_notification_preferences(_uid(user_uid, "user_uid"))
|
|
|
+
|
|
|
+ def replace_notification_preferences(
|
|
|
+ self,
|
|
|
+ payload: Any,
|
|
|
+ *,
|
|
|
+ expected_revision: int,
|
|
|
+ actor_uid: str,
|
|
|
+ ):
|
|
|
+ body = _closed(
|
|
|
+ payload,
|
|
|
+ {"enabled_channels", "subscribed_events", "quiet_hours"},
|
|
|
+ "notification preferences",
|
|
|
+ )
|
|
|
+ channels = sorted(
|
|
|
+ {_string(item, "channel", 20) for item in body.get("enabled_channels", [])}
|
|
|
+ )
|
|
|
+ if not set(channels) <= NOTIFICATION_CHANNELS:
|
|
|
+ raise ValueError("unsupported notification channel")
|
|
|
+ events = sorted(
|
|
|
+ {
|
|
|
+ _string(item, "subscribed event", 80)
|
|
|
+ for item in body.get("subscribed_events", [])
|
|
|
+ }
|
|
|
+ )
|
|
|
+ quiet_hours = body.get("quiet_hours") or {}
|
|
|
+ if not isinstance(quiet_hours, dict):
|
|
|
+ raise ValueError("quiet_hours must be an object")
|
|
|
+ record = {
|
|
|
+ "user_uid": _uid(actor_uid, "actor_uid"),
|
|
|
+ "enabled_channels": channels,
|
|
|
+ "subscribed_events": events,
|
|
|
+ "quiet_hours": copy.deepcopy(quiet_hours),
|
|
|
+ "updated_at": self.now_factory().isoformat(),
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ result = self.repository.replace_notification_preferences(
|
|
|
+ record, int(expected_revision)
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def dashboard(self, *, at: datetime | str | None = None):
|
|
|
+ timestamp = _time(at, "at") if at is not None else self.now_factory()
|
|
|
+ return self.repository.dashboard(timestamp)
|