govoplan-workflow-engine/src/govoplan_workflow_engine/backend/dsar_provider.py

from __future__ import annotations

from collections.abc import Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import or_
from sqlalchemy.orm import Session

from govoplan_core.core.dsar import (
    DsarErasureActionRef,
    DsarExecutionResultRef,
    DsarRecordRef,
    DsarSubjectRef,
    dsar_capability_name,
)
from govoplan_workflow_engine.backend.db.models import (
    WorkflowDefinition,
    WorkflowDefinitionRevision,
    WorkflowInstance,
    WorkflowInstanceEvent,
    WorkflowInstanceStep,
    WorkflowTrigger,
    WorkflowTriggerDelivery,
    WorkflowWaitState,
)


WORKFLOW_ENGINE_DSAR_CAPABILITY = dsar_capability_name("workflow_engine")
_MAX_RECORDS = 5_000
_CONFLICT = object()
_DIRECT_ALIASES = {
    "definition_id": ("workflow_engine.definition", "workflow.definition"),
    "revision_id": (
        "workflow_engine.definition_revision",
        "workflow_engine.revision",
    ),
    "instance_id": ("workflow_engine.instance", "workflow.instance"),
    "step_id": ("workflow_engine.step", "workflow.step"),
    "event_id": ("workflow_engine.event", "workflow.event"),
    "trigger_id": ("workflow_engine.trigger", "workflow.trigger"),
    "delivery_id": (
        "workflow_engine.trigger_delivery",
        "workflow_engine.delivery",
    ),
    "wait_id": ("workflow_engine.wait_state", "workflow_engine.wait"),
}
_RESOURCE_MODELS = {
    "workflow_definition": WorkflowDefinition,
    "workflow_definition_revision": WorkflowDefinitionRevision,
    "workflow_instance": WorkflowInstance,
    "workflow_instance_step": WorkflowInstanceStep,
    "workflow_instance_event": WorkflowInstanceEvent,
    "workflow_trigger": WorkflowTrigger,
    "workflow_trigger_delivery": WorkflowTriggerDelivery,
    "workflow_wait_state": WorkflowWaitState,
}


@dataclass(frozen=True, slots=True)
class _Selectors:
    account_id: str | None
    identity_id: str | None
    membership_id: str | None
    direct: dict[str, str]

    @property
    def subject_ids(self) -> tuple[str, ...]:
        return tuple(
            value
            for value in (self.account_id, self.identity_id, self.membership_id)
            if value
        )


@dataclass(frozen=True, slots=True)
class _Match:
    resource_type: str
    row: Any
    category: str


class WorkflowEngineDsarProvider:
    provider_id = "workflow_engine"
    module_id = "workflow_engine"

    def search_subject(
        self,
        session: object,
        *,
        tenant_id: str,
        subject: DsarSubjectRef,
    ) -> Sequence[DsarRecordRef]:
        db = _session(session)
        selectors = _selectors(subject)
        if selectors is None or not (selectors.subject_ids or selectors.direct):
            return ()
        direct = _direct_matches(db, tenant_id=tenant_id, selectors=selectors)
        if direct is None:
            return ()
        if direct:
            if selectors.subject_ids and not all(
                _correlates(
                    db,
                    tenant_id=tenant_id,
                    match=match,
                    subject_ids=selectors.subject_ids,
                )
                for match in direct
            ):
                return ()
            matches = direct
        else:
            matches = _canonical_matches(
                db,
                tenant_id=tenant_id,
                selectors=selectors,
            )

        records: list[DsarRecordRef] = []
        seen: set[tuple[str, str]] = set()
        for match in matches:
            key = (match.resource_type, str(match.row.id))
            if key in seen:
                continue
            if len(records) >= _MAX_RECORDS:
                raise ValueError(
                    "Workflow Engine DSAR result limit exceeded; narrow the selectors."
                )
            seen.add(key)
            records.append(_record(match))
        return tuple(records)

    def plan_erasure(
        self,
        session: object,
        *,
        tenant_id: str,
        subject: DsarSubjectRef,
        records: Sequence[DsarRecordRef],
    ) -> Sequence[DsarErasureActionRef]:
        del tenant_id
        _session(session)
        if _selectors(subject) is None:
            raise ValueError("Workflow Engine DSAR subject selectors conflict.")
        actions: list[DsarErasureActionRef] = []
        for record in records:
            _validate_record(record)
            kind = _planned_kind(record)
            executable = kind in {"anonymize", "revoke"}
            actions.append(
                DsarErasureActionRef(
                    action_id=(
                        f"workflow_engine:{kind}:{record.resource_type}:"
                        f"{record.resource_id}"
                    ),
                    provider_id=self.provider_id,
                    module_id=self.module_id,
                    kind=kind,
                    resource_type=record.resource_type,
                    resource_id=record.resource_id,
                    title=(
                        f"Minimize {record.title}"
                        if kind == "anonymize"
                        else f"{kind.replace('_', ' ').title()} {record.title}"
                    ),
                    rationale=_rationale(record, kind=kind),
                    executable=executable,
                    irreversible=kind == "anonymize",
                    metadata={"record_category": record.category},
                )
            )
        return tuple(actions)

    def execute_erasure(
        self,
        session: object,
        *,
        tenant_id: str,
        subject: DsarSubjectRef,
        actions: Sequence[DsarErasureActionRef],
        request_id: str,
    ) -> Sequence[DsarExecutionResultRef]:
        db = _session(session)
        selectors = _selectors(subject)
        if selectors is None:
            raise ValueError("Workflow Engine DSAR subject selectors conflict.")
        results: list[DsarExecutionResultRef] = []
        for action in actions:
            _validate_action(action)
            if not action.executable:
                results.append(
                    DsarExecutionResultRef(
                        action_id=action.action_id,
                        status="blocked",
                        summary=(
                            "Review live work, third parties, decision evidence, "
                            "retention, and downstream effects before changing state."
                        ),
                        evidence={"request_id": request_id},
                    )
                )
                continue
            model = _RESOURCE_MODELS[action.resource_type]
            row = (
                db.query(model)
                .filter(model.tenant_id == tenant_id, model.id == action.resource_id)
                .with_for_update()
                .one_or_none()
            )
            if row is None:
                status = "unchanged"
                summary = "Workflow row was already absent or minimized."
            else:
                match = _Match(action.resource_type, row, "execution")
                if not (
                    _directly_targets(selectors, match)
                    or _correlates(
                        db,
                        tenant_id=tenant_id,
                        match=match,
                        subject_ids=selectors.subject_ids,
                    )
                    or _already_revoked(action.resource_type, row)
                ):
                    raise ValueError(
                        "Workflow Engine DSAR action is not corroborated by the subject."
                    )
                status, summary = _execute_action(
                    db,
                    resource_type=action.resource_type,
                    row=row,
                    kind=action.kind,
                )
            results.append(
                DsarExecutionResultRef(
                    action_id=action.action_id,
                    status=status,
                    summary=summary,
                    evidence={"request_id": request_id},
                )
            )
        return tuple(results)


def _direct_matches(
    session: Session,
    *,
    tenant_id: str,
    selectors: _Selectors,
) -> list[_Match] | None:
    matches: list[_Match] = []
    for selector, raw_value in selectors.direct.items():
        value = _strip_prefix(raw_value)
        if selector == "definition_id":
            definition = _one(session, WorkflowDefinition, tenant_id, value)
            if definition is None:
                return None
            current = _definition_package(
                session,
                tenant_id=tenant_id,
                definition=definition,
            )
        elif selector == "instance_id":
            instance = _one(session, WorkflowInstance, tenant_id, value)
            if instance is None:
                return None
            current = _instance_package(
                session,
                tenant_id=tenant_id,
                instance=instance,
            )
        else:
            model, resource_type = {
                "revision_id": (
                    WorkflowDefinitionRevision,
                    "workflow_definition_revision",
                ),
                "step_id": (WorkflowInstanceStep, "workflow_instance_step"),
                "event_id": (WorkflowInstanceEvent, "workflow_instance_event"),
                "trigger_id": (WorkflowTrigger, "workflow_trigger"),
                "delivery_id": (
                    WorkflowTriggerDelivery,
                    "workflow_trigger_delivery",
                ),
                "wait_id": (WorkflowWaitState, "workflow_wait_state"),
            }[selector]
            row = _one(session, model, tenant_id, value)
            if row is None:
                return None
            current = [_Match(resource_type, row, _direct_category(resource_type, row))]
        matches.extend(current)
        if len(matches) > _MAX_RECORDS:
            raise ValueError(
                "Workflow Engine DSAR result limit exceeded; narrow the selectors."
            )
    return matches


def _definition_package(
    session: Session,
    *,
    tenant_id: str,
    definition: WorkflowDefinition,
) -> list[_Match]:
    matches = [_Match("workflow_definition", definition, "workflow_definition_package")]
    specs = (
        (WorkflowDefinitionRevision, "workflow_definition_revision"),
        (WorkflowInstance, "workflow_instance"),
        (WorkflowTrigger, "workflow_trigger"),
        (WorkflowTriggerDelivery, "workflow_trigger_delivery"),
    )
    instances: list[WorkflowInstance] = []
    for model, resource_type in specs:
        rows = (
            session.query(model)
            .filter(model.tenant_id == tenant_id, model.definition_id == definition.id)
            .order_by(model.id)
            .limit(_MAX_RECORDS + 1)
            .all()
        )
        if model is WorkflowInstance:
            instances = rows
        matches.extend(
            _Match(resource_type, row, "workflow_definition_package") for row in rows
        )
    instance_ids = {row.id for row in instances}
    if instance_ids:
        for model, resource_type in (
            (WorkflowInstanceStep, "workflow_instance_step"),
            (WorkflowInstanceEvent, "workflow_instance_event"),
            (WorkflowWaitState, "workflow_wait_state"),
        ):
            rows = (
                session.query(model)
                .filter(
                    model.tenant_id == tenant_id,
                    model.instance_id.in_(instance_ids),
                )
                .order_by(model.id)
                .limit(_MAX_RECORDS + 1)
                .all()
            )
            matches.extend(
                _Match(resource_type, row, "workflow_definition_package")
                for row in rows
            )
    return matches


def _instance_package(
    session: Session,
    *,
    tenant_id: str,
    instance: WorkflowInstance,
) -> list[_Match]:
    matches = [_Match("workflow_instance", instance, "workflow_instance_package")]
    specs = (
        (WorkflowInstanceStep, "workflow_instance_step"),
        (WorkflowInstanceEvent, "workflow_instance_event"),
        (WorkflowWaitState, "workflow_wait_state"),
    )
    for model, resource_type in specs:
        rows = (
            session.query(model)
            .filter(model.tenant_id == tenant_id, model.instance_id == instance.id)
            .order_by(model.id)
            .limit(_MAX_RECORDS + 1)
            .all()
        )
        matches.extend(
            _Match(resource_type, row, "workflow_instance_package") for row in rows
        )
    return matches


def _canonical_matches(
    session: Session,
    *,
    tenant_id: str,
    selectors: _Selectors,
) -> list[_Match]:
    subject_ids = selectors.subject_ids
    if not subject_ids:
        return []
    triggers = (
        session.query(WorkflowTrigger)
        .filter(
            WorkflowTrigger.tenant_id == tenant_id,
            or_(
                WorkflowTrigger.authorization_account_id.in_(subject_ids),
                WorkflowTrigger.authorization_membership_id.in_(subject_ids),
            ),
        )
        .order_by(WorkflowTrigger.id)
        .limit(_MAX_RECORDS + 1)
        .all()
    )
    matches = [
        _Match("workflow_trigger", row, "workflow_automation_authority")
        for row in triggers
    ]
    instances = (
        session.query(WorkflowInstance)
        .filter(WorkflowInstance.tenant_id == tenant_id)
        .order_by(WorkflowInstance.id)
        .limit(_MAX_RECORDS + 1)
        .all()
    )
    if len(instances) > _MAX_RECORDS:
        raise ValueError(
            "Workflow Engine DSAR instance scan limit exceeded; use an exact instance reference."
        )
    for row in instances:
        authorization = dict(row.authorization_ or {})
        if any(
            str(authorization.get(key) or "") == value
            for key, value in (
                ("account_id", selectors.account_id),
                ("identity_id", selectors.identity_id),
                ("membership_id", selectors.membership_id),
            )
            if value
        ):
            matches.append(
                _Match(
                    "workflow_instance",
                    row,
                    "workflow_subject_authorization",
                )
            )
    steps = (
        session.query(WorkflowInstanceStep)
        .filter(
            WorkflowInstanceStep.tenant_id == tenant_id,
            WorkflowInstanceStep.work_assignment_id.in_(subject_ids),
        )
        .order_by(WorkflowInstanceStep.id)
        .limit(_MAX_RECORDS + 1)
        .all()
    )
    matches.extend(
        _Match("workflow_instance_step", row, "workflow_subject_work_assignment")
        for row in steps
    )
    if len(matches) > _MAX_RECORDS:
        raise ValueError(
            "Workflow Engine DSAR result limit exceeded; narrow the selectors."
        )
    specs = (
        (
            WorkflowDefinition,
            or_(
                WorkflowDefinition.created_by.in_(subject_ids),
                WorkflowDefinition.updated_by.in_(subject_ids),
            ),
            "workflow_definition",
        ),
        (
            WorkflowDefinitionRevision,
            WorkflowDefinitionRevision.created_by.in_(subject_ids),
            "workflow_definition_revision",
        ),
        (
            WorkflowInstance,
            WorkflowInstance.created_by.in_(subject_ids),
            "workflow_instance",
        ),
        (
            WorkflowInstanceStep,
            WorkflowInstanceStep.completed_by.in_(subject_ids),
            "workflow_instance_step",
        ),
        (
            WorkflowInstanceEvent,
            WorkflowInstanceEvent.actor_id.in_(subject_ids),
            "workflow_instance_event",
        ),
        (
            WorkflowTrigger,
            or_(
                WorkflowTrigger.created_by.in_(subject_ids),
                WorkflowTrigger.updated_by.in_(subject_ids),
            ),
            "workflow_trigger",
        ),
    )
    for model, condition, resource_type in specs:
        rows = (
            session.query(model)
            .filter(model.tenant_id == tenant_id, condition)
            .order_by(model.id)
            .limit(_MAX_RECORDS + 1)
            .all()
        )
        matches.extend(
            _Match(resource_type, row, "workflow_operator_attribution") for row in rows
        )
        if len(matches) > _MAX_RECORDS:
            raise ValueError(
                "Workflow Engine DSAR result limit exceeded; narrow the selectors."
            )
    return matches


def _one(session: Session, model: Any, tenant_id: str, row_id: str) -> Any | None:
    return (
        session.query(model)
        .filter(model.tenant_id == tenant_id, model.id == row_id)
        .one_or_none()
    )


def _direct_category(resource_type: str, row: Any) -> str:
    if resource_type == "workflow_trigger_delivery":
        return (
            "terminal_workflow_delivery"
            if row.status in {"succeeded", "failed", "skipped", "cancelled"}
            else "active_workflow_delivery"
        )
    return "workflow_governed_state"


def _correlates(
    session: Session,
    *,
    tenant_id: str,
    match: _Match,
    subject_ids: tuple[str, ...],
) -> bool:
    row = match.row
    if any(
        str(getattr(row, field, "") or "") in subject_ids
        for field in (
            "created_by",
            "updated_by",
            "completed_by",
            "actor_id",
            "work_assignment_id",
            "authorization_account_id",
            "authorization_membership_id",
        )
    ):
        return True
    if match.resource_type == "workflow_instance":
        authorization = dict(row.authorization_ or {})
        if any(
            str(authorization.get(key) or "") in subject_ids
            for key in ("account_id", "identity_id", "membership_id")
        ):
            return True
    instance_id = getattr(row, "instance_id", None)
    if instance_id:
        instance = session.get(WorkflowInstance, instance_id)
        if instance and instance.tenant_id == tenant_id:
            return _correlates(
                session,
                tenant_id=tenant_id,
                match=_Match("workflow_instance", instance, "related"),
                subject_ids=subject_ids,
            )
    trigger_id = getattr(row, "trigger_id", None)
    if trigger_id:
        trigger = session.get(WorkflowTrigger, trigger_id)
        if trigger and trigger.tenant_id == tenant_id:
            return _correlates(
                session,
                tenant_id=tenant_id,
                match=_Match("workflow_trigger", trigger, "related"),
                subject_ids=subject_ids,
            )
    definition_id = getattr(row, "definition_id", None)
    if definition_id:
        definition = session.get(WorkflowDefinition, definition_id)
        return bool(
            definition
            and definition.tenant_id == tenant_id
            and (
                definition.created_by in subject_ids
                or definition.updated_by in subject_ids
            )
        )
    return False


def _directly_targets(selectors: _Selectors, match: _Match) -> bool:
    selector = {
        "workflow_definition": "definition_id",
        "workflow_definition_revision": "revision_id",
        "workflow_instance": "instance_id",
        "workflow_instance_step": "step_id",
        "workflow_instance_event": "event_id",
        "workflow_trigger": "trigger_id",
        "workflow_trigger_delivery": "delivery_id",
        "workflow_wait_state": "wait_id",
    }[match.resource_type]
    return _strip_prefix(selectors.direct.get(selector, "")) == str(match.row.id)


def _record(match: _Match) -> DsarRecordRef:
    row = match.row
    immutable = match.category == "workflow_operator_attribution"
    return DsarRecordRef(
        provider_id="workflow_engine",
        module_id="workflow_engine",
        resource_type=match.resource_type,
        resource_id=str(row.id),
        category=match.category,
        title=match.resource_type.removeprefix("workflow_").replace("_", " ").title(),
        data={
            key: value
            for key, value in _record_data(match.resource_type, row).items()
            if value is not None
        },
        observed_at=_observed_at(row),
        immutable_evidence=immutable,
        retention_reason=(
            "Institutional Workflow authorship, assignments, transitions, and "
            "decision events remain attributable for governance and audit."
            if immutable
            else None
        ),
        source_path="/workflow",
    )


def _record_data(resource_type: str, row: Any) -> dict[str, object]:
    if resource_type == "workflow_definition":
        return {
            "scope_type": row.scope_type,
            "definition_kind": row.definition_kind,
            "status": row.status,
            "current_revision": row.current_revision,
            "active_revision": row.active_revision,
            "allow_start": row.allow_start,
            "allow_reuse": row.allow_reuse,
            "allow_automation": row.allow_automation,
            "deleted_at": _iso(row.deleted_at),
        }
    if resource_type == "workflow_definition_revision":
        return {
            "revision": row.revision,
            "schema_version": row.schema_version,
            "execution_mode": row.execution_mode,
            "bpmn_runtime_kind": row.bpmn_runtime_kind,
            "bpmn_executable": row.bpmn_executable,
            "created_at": _iso(row.created_at),
        }
    if resource_type == "workflow_instance":
        return {
            "status": row.status,
            "start_origin": row.start_origin,
            "started_at": _iso(row.started_at),
            "finished_at": _iso(row.finished_at),
            "cancellation_requested_at": _iso(row.cancellation_requested_at),
        }
    if resource_type == "workflow_instance_step":
        return {
            "sequence": row.sequence,
            "node_type": row.node_type,
            "status": row.status,
            "attempt": row.attempt,
            "work_assignment_kind": row.work_assignment_kind,
            "work_due_at": _iso(row.work_due_at),
            "started_at": _iso(row.started_at),
            "finished_at": _iso(row.finished_at),
        }
    if resource_type == "workflow_instance_event":
        return {
            "sequence": row.sequence,
            "kind": row.kind,
            "created_at": _iso(row.created_at),
        }
    if resource_type == "workflow_trigger":
        return {
            "kind": row.kind,
            "status": row.status,
            "event_type": row.event_type,
            "authorization_subject_kind": row.authorization_subject_kind,
            "next_fire_at": _iso(row.next_fire_at),
            "last_fire_at": _iso(row.last_fire_at),
        }
    if resource_type == "workflow_trigger_delivery":
        return {
            "invocation_kind": row.invocation_kind,
            "status": row.status,
            "attempts": row.attempts,
            "scheduled_for": _iso(row.scheduled_for),
            "created_at": _iso(row.created_at),
        }
    return {
        "mode": row.mode,
        "status": row.status,
        "due_at": _iso(row.due_at),
        "event_type": row.event_type,
        "revision": row.revision,
        "created_at": _iso(row.created_at),
        "updated_at": _iso(row.updated_at),
    }


def _planned_kind(record: DsarRecordRef) -> str:
    if record.category == "terminal_workflow_delivery":
        return "anonymize"
    if record.category == "workflow_automation_authority":
        return "revoke"
    if record.category == "workflow_operator_attribution":
        return "retain"
    return "manual_review"


def _rationale(record: DsarRecordRef, *, kind: str) -> str:
    if kind == "anonymize":
        return (
            "Clear terminal trigger event and error detail while preserving the "
            "replay identity and minimal delivery evidence."
        )
    if kind == "revoke":
        return (
            "Disable automation and remove subject-linked delegated authority "
            "without deleting historical workflow evidence."
        )
    if kind == "retain":
        return record.retention_reason or "Retain institutional attribution evidence."
    return (
        "A Workflow owner must review live work, assignments, third parties, "
        "decision evidence, source authority, and retention."
    )


def _execute_action(
    session: Session,
    *,
    resource_type: str,
    row: Any,
    kind: str,
) -> tuple[str, str]:
    if resource_type == "workflow_trigger_delivery" and kind == "anonymize":
        if row.status not in {"succeeded", "failed", "skipped", "cancelled"}:
            raise ValueError("Active Workflow deliveries require manual review.")
        changed = _replace_fields(row, {"event_": None, "error": None})
    elif resource_type == "workflow_trigger" and kind == "revoke":
        changed = _replace_fields(
            row,
            {
                "status": "disabled",
                "config_": {},
                "last_error": None,
                "authorization_account_id": None,
                "authorization_membership_id": None,
                "authorization_service_account_id": None,
                "authorization_ref": f"revoked:{row.id}",
                "grant_scopes": [],
            },
        )
    else:
        raise ValueError("Workflow Engine DSAR executable action is unsupported.")
    if changed:
        session.flush()
        return "executed", "Personal Workflow detail minimized."
    return "unchanged", "Personal Workflow detail was already minimized."


def _replace_fields(row: Any, values: dict[str, object]) -> bool:
    changed = False
    for field, value in values.items():
        if getattr(row, field) != value:
            setattr(row, field, value)
            changed = True
    return changed


def _already_revoked(resource_type: str, row: Any) -> bool:
    return bool(
        resource_type == "workflow_trigger"
        and row.status == "disabled"
        and row.authorization_account_id is None
        and row.authorization_membership_id is None
        and row.authorization_service_account_id is None
        and row.authorization_ref == f"revoked:{row.id}"
    )


def _selectors(subject: DsarSubjectRef) -> _Selectors | None:
    refs = subject.external_references
    canonical = (
        _coalesce(
            subject.account_id,
            refs.get("workflow_engine.account"),
            refs.get("access.account"),
        ),
        _coalesce(
            subject.identity_id,
            refs.get("workflow_engine.identity"),
            refs.get("identity.id"),
        ),
        _coalesce(
            subject.membership_id,
            refs.get("workflow_engine.membership"),
            refs.get("tenancy.membership"),
        ),
    )
    if any(value is _CONFLICT for value in canonical):
        return None
    direct: dict[str, str] = {}
    for selector, aliases in _DIRECT_ALIASES.items():
        value = _coalesce(*(refs.get(alias) for alias in aliases))
        if value is _CONFLICT:
            return None
        if value:
            direct[selector] = str(value)
    return _Selectors(
        account_id=_optional(canonical[0]),
        identity_id=_optional(canonical[1]),
        membership_id=_optional(canonical[2]),
        direct=direct,
    )


def _coalesce(*values: str | None) -> str | None | object:
    normalized = {str(value).strip() for value in values if str(value or "").strip()}
    if len(normalized) > 1:
        return _CONFLICT
    return next(iter(normalized), None)


def _optional(value: object) -> str | None:
    return value if isinstance(value, str) and value else None


def _strip_prefix(value: str) -> str:
    return value.partition(":")[2] if ":" in value else value


def _observed_at(row: Any) -> datetime | None:
    for field in ("finished_at", "created_at", "updated_at", "started_at"):
        value = getattr(row, field, None)
        if isinstance(value, datetime):
            return _aware(value)
    return None


def _iso(value: datetime | None) -> str | None:
    aware = _aware(value)
    return aware.isoformat() if aware else None


def _aware(value: datetime | None) -> datetime | None:
    if value is None or value.tzinfo is not None:
        return value
    return value.replace(tzinfo=timezone.utc)


def _session(value: object) -> Session:
    if not isinstance(value, Session):
        raise TypeError("Workflow Engine DSAR requires a SQLAlchemy Session.")
    return value


def _validate_record(record: DsarRecordRef) -> None:
    if record.provider_id != "workflow_engine" or record.module_id != "workflow_engine":
        raise ValueError("Workflow Engine DSAR cannot plan a foreign provider record.")
    if record.resource_type not in _RESOURCE_MODELS or not record.resource_id:
        raise ValueError("Workflow Engine DSAR record identity is invalid.")


def _validate_action(action: DsarErasureActionRef) -> None:
    if action.provider_id != "workflow_engine" or action.module_id != "workflow_engine":
        raise ValueError(
            "Workflow Engine DSAR cannot execute a foreign provider action."
        )
    if action.resource_type not in _RESOURCE_MODELS or not action.action_id.startswith(
        "workflow_engine:"
    ):
        raise ValueError("Workflow Engine DSAR action identity is invalid.")


__all__ = ["WORKFLOW_ENGINE_DSAR_CAPABILITY", "WorkflowEngineDsarProvider"]