From 6737b60c113732875da7766f9c95376be261b609 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 28 Jul 2026 13:48:06 +0200 Subject: [PATCH] feat: persist and edit versioned workflow definitions --- README.md | 20 +- src/govoplan_workflow/backend/db/__init__.py | 9 + src/govoplan_workflow/backend/db/models.py | 126 ++++ src/govoplan_workflow/backend/manifest.py | 71 +- .../backend/migrations/__init__.py | 1 + .../backend/migrations/versions/__init__.py | 1 + ...a7c4e2f9b1d3_v0114_workflow_definitions.py | 201 +++++ src/govoplan_workflow/backend/router.py | 338 +++++++++ src/govoplan_workflow/backend/schemas.py | 72 ++ src/govoplan_workflow/backend/service.py | 399 ++++++++++ tests/test_manifest.py | 11 + tests/test_migrations.py | 48 ++ tests/test_service.py | 225 ++++++ webui/package.json | 32 + webui/src/api/workflow.ts | 158 ++++ .../src/features/workflow/WorkflowCanvas.tsx | 229 ++++++ .../features/workflow/WorkflowInspector.tsx | 185 +++++ webui/src/features/workflow/WorkflowNode.tsx | 99 +++ webui/src/features/workflow/WorkflowPage.tsx | 703 +++++++++++++++++ webui/src/features/workflow/model.ts | 193 +++++ webui/src/index.ts | 2 + webui/src/module.ts | 45 ++ webui/src/styles/workflow.css | 705 ++++++++++++++++++ webui/src/vite-env.d.ts | 16 + webui/tsconfig.json | 33 + 25 files changed, 3913 insertions(+), 9 deletions(-) create mode 100644 src/govoplan_workflow/backend/db/__init__.py create mode 100644 src/govoplan_workflow/backend/db/models.py create mode 100644 src/govoplan_workflow/backend/migrations/__init__.py create mode 100644 src/govoplan_workflow/backend/migrations/versions/__init__.py create mode 100644 src/govoplan_workflow/backend/migrations/versions/a7c4e2f9b1d3_v0114_workflow_definitions.py create mode 100644 src/govoplan_workflow/backend/service.py create mode 100644 tests/test_migrations.py create mode 100644 tests/test_service.py create mode 100644 webui/package.json create mode 100644 webui/src/api/workflow.ts create mode 100644 webui/src/features/workflow/WorkflowCanvas.tsx create mode 100644 webui/src/features/workflow/WorkflowInspector.tsx create mode 100644 webui/src/features/workflow/WorkflowNode.tsx create mode 100644 webui/src/features/workflow/WorkflowPage.tsx create mode 100644 webui/src/features/workflow/model.ts create mode 100644 webui/src/index.ts create mode 100644 webui/src/module.ts create mode 100644 webui/src/styles/workflow.css create mode 100644 webui/src/vite-env.d.ts create mode 100644 webui/tsconfig.json diff --git a/README.md b/README.md index 55a9c2d..0b17642 100644 --- a/README.md +++ b/README.md @@ -11,13 +11,19 @@ between modules without importing their implementations. It coordinates cases, tasks, forms, files, templates, mail, appointments, payments, and records through capabilities, events, commands, and DTOs. -The first executable slice exposes a workflow-specific node library and graph -validation API through `/workflow/node-types` and -`/workflow/definitions/validate`. It uses Core's domain-neutral definition graph -contract, but applies Workflow constraints: exactly one trigger, at least one -outcome, governed configuration fields, connected nodes, and permitted loops for -correction and retry paths. Dataflow uses the same graph contract with its own -acyclic transformation library. +Workflow exposes a workflow-specific node library, a full definition editor, +validation APIs, tenant-isolated definitions, and immutable graph revisions. +Activation pins the exact revision that future instances will execute. It uses +Core's domain-neutral definition graph contract, but applies Workflow +constraints: exactly one trigger, at least one outcome, governed configuration +fields, connected nodes, and permitted loops for correction and retry paths. +Dataflow uses the same graph contract with its own acyclic transformation +library. + +The current slice deliberately stops before executing process instances. +Instance state, resumable transitions, human activities, retries, and event +subscriptions must build on the pinned definition and versioned module +capabilities rather than bypassing them. See [docs/CONCEPT.md](docs/CONCEPT.md) for the complete module concept. diff --git a/src/govoplan_workflow/backend/db/__init__.py b/src/govoplan_workflow/backend/db/__init__.py new file mode 100644 index 0000000..58df07b --- /dev/null +++ b/src/govoplan_workflow/backend/db/__init__.py @@ -0,0 +1,9 @@ +from govoplan_workflow.backend.db.models import ( + WorkflowDefinition, + WorkflowDefinitionRevision, +) + +__all__ = [ + "WorkflowDefinition", + "WorkflowDefinitionRevision", +] diff --git a/src/govoplan_workflow/backend/db/models.py b/src/govoplan_workflow/backend/db/models.py new file mode 100644 index 0000000..44ca2d3 --- /dev/null +++ b/src/govoplan_workflow/backend/db/models.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +import uuid +from datetime import datetime +from typing import Any + +from sqlalchemy import ( + DateTime, + ForeignKey, + Index, + Integer, + JSON, + String, + Text, + UniqueConstraint, +) +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from govoplan_core.db.base import Base, TimestampMixin + + +def new_uuid() -> str: + return str(uuid.uuid4()) + + +class WorkflowDefinition(Base, TimestampMixin): + __tablename__ = "workflow_definitions" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "definition_key", + name="uq_workflow_definition_key", + ), + Index("ix_workflow_definitions_tenant_status", "tenant_id", "status"), + Index("ix_workflow_definitions_tenant_updated", "tenant_id", "updated_at"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + definition_key: Mapped[str] = mapped_column(String(120), nullable=False) + name: Mapped[str] = mapped_column(String(300), nullable=False) + description: Mapped[str | None] = mapped_column(Text) + status: Mapped[str] = mapped_column( + String(32), + default="draft", + nullable=False, + index=True, + ) + current_revision: Mapped[int] = mapped_column(Integer, default=1, nullable=False) + active_revision: Mapped[int | None] = mapped_column(Integer, nullable=True) + metadata_: Mapped[dict[str, Any]] = mapped_column( + "metadata", + JSON, + default=dict, + nullable=False, + ) + created_by: Mapped[str | None] = mapped_column( + String(255), + nullable=True, + index=True, + ) + updated_by: Mapped[str | None] = mapped_column( + String(255), + nullable=True, + index=True, + ) + deleted_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + index=True, + ) + + revisions: Mapped[list["WorkflowDefinitionRevision"]] = relationship( + back_populates="definition", + cascade="all, delete-orphan", + order_by="WorkflowDefinitionRevision.revision", + ) + + +class WorkflowDefinitionRevision(Base, TimestampMixin): + __tablename__ = "workflow_definition_revisions" + __table_args__ = ( + UniqueConstraint( + "definition_id", + "revision", + name="uq_workflow_definition_revision", + ), + Index( + "ix_workflow_definition_revisions_tenant_definition", + "tenant_id", + "definition_id", + ), + Index( + "ix_workflow_definition_revisions_content_hash", + "tenant_id", + "content_hash", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + definition_id: Mapped[str] = mapped_column( + ForeignKey("workflow_definitions.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + revision: Mapped[int] = mapped_column(Integer, nullable=False) + schema_version: Mapped[int] = mapped_column(Integer, default=1, nullable=False) + graph: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + content_hash: Mapped[str] = mapped_column(String(64), nullable=False) + library_id: Mapped[str] = mapped_column(String(100), nullable=False) + library_version: Mapped[str] = mapped_column(String(40), nullable=False) + created_by: Mapped[str | None] = mapped_column( + String(255), + nullable=True, + index=True, + ) + + definition: Mapped[WorkflowDefinition] = relationship(back_populates="revisions") + + +__all__ = [ + "WorkflowDefinition", + "WorkflowDefinitionRevision", + "new_uuid", +] diff --git a/src/govoplan_workflow/backend/manifest.py b/src/govoplan_workflow/backend/manifest.py index 6d7c65e..62489cb 100644 --- a/src/govoplan_workflow/backend/manifest.py +++ b/src/govoplan_workflow/backend/manifest.py @@ -1,17 +1,30 @@ from __future__ import annotations +from pathlib import Path + from govoplan_core.core.access import ( CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER, ) +from govoplan_core.core.dataflows import CAPABILITY_DATAFLOW_RUN_LIFECYCLE +from govoplan_core.core.module_guards import ( + drop_table_retirement_provider, + persistent_table_uninstall_guard, +) from govoplan_core.core.modules import ( DocumentationTopic, + FrontendModule, + MigrationSpec, ModuleContext, ModuleInterfaceProvider, + ModuleInterfaceRequirement, ModuleManifest, + NavItem, PermissionDefinition, RoleTemplate, ) +from govoplan_core.db.base import Base +from govoplan_workflow.backend.db import models as workflow_models MODULE_ID = "workflow" @@ -117,14 +130,68 @@ manifest = ModuleManifest( optional_capabilities=( CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, + CAPABILITY_DATAFLOW_RUN_LIFECYCLE, ), provides_interfaces=( ModuleInterfaceProvider(name="workflow.definition_graph", version="0.1.0"), ModuleInterfaceProvider(name="workflow.node_library", version="0.1.0"), + ModuleInterfaceProvider(name="workflow.definition_catalogue", version="0.1.0"), + ), + requires_interfaces=( + ModuleInterfaceRequirement( + name="dataflow.run_lifecycle", + version_min="0.1.14", + version_max_exclusive="1.0.0", + optional=True, + ), ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, + nav_items=( + NavItem( + path="/workflow", + label="Workflow", + icon="workflow", + required_any=(DEFINITION_READ_SCOPE, ADMIN_SCOPE), + order=74, + ), + ), + frontend=FrontendModule( + module_id=MODULE_ID, + package_name="@govoplan/workflow-webui", + nav_items=( + NavItem( + path="/workflow", + label="Workflow", + icon="workflow", + required_any=(DEFINITION_READ_SCOPE, ADMIN_SCOPE), + order=74, + ), + ), + ), route_factory=_router, + migration_spec=MigrationSpec( + module_id=MODULE_ID, + metadata=Base.metadata, + script_location=str(Path(__file__).with_name("migrations") / "versions"), + retirement_supported=True, + retirement_provider=drop_table_retirement_provider( + workflow_models.WorkflowDefinitionRevision, + workflow_models.WorkflowDefinition, + label="Workflow", + ), + retirement_notes=( + "Destructive retirement drops Workflow definitions and immutable revisions " + "after the installer captures a database snapshot." + ), + ), + uninstall_guard_providers=( + persistent_table_uninstall_guard( + workflow_models.WorkflowDefinition, + workflow_models.WorkflowDefinitionRevision, + label="Workflow", + ), + ), documentation=( DocumentationTopic( id="workflow.definition-graphs", @@ -135,8 +202,8 @@ manifest = ModuleManifest( "and outcome node library on top of Core's domain-neutral graph contract. " "Unlike Dataflow, Workflow permits cycles for correction and retry paths. " "Module actions are addressed through versioned capabilities rather than " - "implementation imports. Definition persistence and execution build on this " - "validated contract in subsequent slices." + "implementation imports. Definitions are persisted as immutable graph " + "revisions; activation pins the exact revision used by future instances." ), layer="available", documentation_types=("admin", "user"), diff --git a/src/govoplan_workflow/backend/migrations/__init__.py b/src/govoplan_workflow/backend/migrations/__init__.py new file mode 100644 index 0000000..863565f --- /dev/null +++ b/src/govoplan_workflow/backend/migrations/__init__.py @@ -0,0 +1 @@ +"""Workflow database migrations.""" diff --git a/src/govoplan_workflow/backend/migrations/versions/__init__.py b/src/govoplan_workflow/backend/migrations/versions/__init__.py new file mode 100644 index 0000000..7dd88e5 --- /dev/null +++ b/src/govoplan_workflow/backend/migrations/versions/__init__.py @@ -0,0 +1 @@ +"""Workflow Alembic revisions.""" diff --git a/src/govoplan_workflow/backend/migrations/versions/a7c4e2f9b1d3_v0114_workflow_definitions.py b/src/govoplan_workflow/backend/migrations/versions/a7c4e2f9b1d3_v0114_workflow_definitions.py new file mode 100644 index 0000000..bfa46c8 --- /dev/null +++ b/src/govoplan_workflow/backend/migrations/versions/a7c4e2f9b1d3_v0114_workflow_definitions.py @@ -0,0 +1,201 @@ +"""v0.1.14 Workflow definitions + +Revision ID: a7c4e2f9b1d3 +Revises: None +Create Date: 2026-07-28 00:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "a7c4e2f9b1d3" +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "workflow_definitions", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("definition_key", sa.String(length=120), nullable=False), + sa.Column("name", sa.String(length=300), nullable=False), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("status", sa.String(length=32), nullable=False), + sa.Column("current_revision", sa.Integer(), nullable=False), + sa.Column("active_revision", sa.Integer(), nullable=True), + sa.Column("metadata", sa.JSON(), nullable=False), + sa.Column("created_by", sa.String(length=255), nullable=True), + sa.Column("updated_by", sa.String(length=255), nullable=True), + sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id", name=op.f("pk_workflow_definitions")), + sa.UniqueConstraint( + "tenant_id", + "definition_key", + name="uq_workflow_definition_key", + ), + ) + op.create_index( + op.f("ix_workflow_definitions_created_by"), + "workflow_definitions", + ["created_by"], + unique=False, + ) + op.create_index( + op.f("ix_workflow_definitions_deleted_at"), + "workflow_definitions", + ["deleted_at"], + unique=False, + ) + op.create_index( + op.f("ix_workflow_definitions_status"), + "workflow_definitions", + ["status"], + unique=False, + ) + op.create_index( + op.f("ix_workflow_definitions_tenant_id"), + "workflow_definitions", + ["tenant_id"], + unique=False, + ) + op.create_index( + op.f("ix_workflow_definitions_updated_by"), + "workflow_definitions", + ["updated_by"], + unique=False, + ) + op.create_index( + "ix_workflow_definitions_tenant_status", + "workflow_definitions", + ["tenant_id", "status"], + unique=False, + ) + op.create_index( + "ix_workflow_definitions_tenant_updated", + "workflow_definitions", + ["tenant_id", "updated_at"], + unique=False, + ) + + op.create_table( + "workflow_definition_revisions", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("definition_id", sa.String(length=36), nullable=False), + sa.Column("revision", sa.Integer(), nullable=False), + sa.Column("schema_version", sa.Integer(), nullable=False), + sa.Column("graph", sa.JSON(), nullable=False), + sa.Column("content_hash", sa.String(length=64), nullable=False), + sa.Column("library_id", sa.String(length=100), nullable=False), + sa.Column("library_version", sa.String(length=40), nullable=False), + sa.Column("created_by", sa.String(length=255), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["definition_id"], + ["workflow_definitions.id"], + name=op.f( + "fk_workflow_definition_revisions_definition_id_workflow_definitions" + ), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint( + "id", + name=op.f("pk_workflow_definition_revisions"), + ), + sa.UniqueConstraint( + "definition_id", + "revision", + name="uq_workflow_definition_revision", + ), + ) + op.create_index( + op.f("ix_workflow_definition_revisions_created_by"), + "workflow_definition_revisions", + ["created_by"], + unique=False, + ) + op.create_index( + op.f("ix_workflow_definition_revisions_definition_id"), + "workflow_definition_revisions", + ["definition_id"], + unique=False, + ) + op.create_index( + op.f("ix_workflow_definition_revisions_tenant_id"), + "workflow_definition_revisions", + ["tenant_id"], + unique=False, + ) + op.create_index( + "ix_workflow_definition_revisions_content_hash", + "workflow_definition_revisions", + ["tenant_id", "content_hash"], + unique=False, + ) + op.create_index( + "ix_workflow_definition_revisions_tenant_definition", + "workflow_definition_revisions", + ["tenant_id", "definition_id"], + unique=False, + ) + + +def downgrade() -> None: + op.drop_index( + "ix_workflow_definition_revisions_tenant_definition", + table_name="workflow_definition_revisions", + ) + op.drop_index( + "ix_workflow_definition_revisions_content_hash", + table_name="workflow_definition_revisions", + ) + op.drop_index( + op.f("ix_workflow_definition_revisions_tenant_id"), + table_name="workflow_definition_revisions", + ) + op.drop_index( + op.f("ix_workflow_definition_revisions_definition_id"), + table_name="workflow_definition_revisions", + ) + op.drop_index( + op.f("ix_workflow_definition_revisions_created_by"), + table_name="workflow_definition_revisions", + ) + op.drop_table("workflow_definition_revisions") + + op.drop_index( + "ix_workflow_definitions_tenant_updated", + table_name="workflow_definitions", + ) + op.drop_index( + "ix_workflow_definitions_tenant_status", + table_name="workflow_definitions", + ) + op.drop_index( + op.f("ix_workflow_definitions_updated_by"), + table_name="workflow_definitions", + ) + op.drop_index( + op.f("ix_workflow_definitions_tenant_id"), + table_name="workflow_definitions", + ) + op.drop_index( + op.f("ix_workflow_definitions_status"), + table_name="workflow_definitions", + ) + op.drop_index( + op.f("ix_workflow_definitions_deleted_at"), + table_name="workflow_definitions", + ) + op.drop_index( + op.f("ix_workflow_definitions_created_by"), + table_name="workflow_definitions", + ) + op.drop_table("workflow_definitions") diff --git a/src/govoplan_workflow/backend/router.py b/src/govoplan_workflow/backend/router.py index 7fddf11..ff5b231 100644 --- a/src/govoplan_workflow/backend/router.py +++ b/src/govoplan_workflow/backend/router.py @@ -1,8 +1,11 @@ from __future__ import annotations from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.orm import Session +from govoplan_core.audit.logging import audit_event from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope +from govoplan_core.db.session import get_session from govoplan_workflow.backend.manifest import ( ADMIN_SCOPE, DEFINITION_READ_SCOPE, @@ -11,6 +14,14 @@ from govoplan_workflow.backend.manifest import ( from govoplan_workflow.backend.node_library import WORKFLOW_GRAPH_LIBRARY from govoplan_workflow.backend.schemas import ( WorkflowConfigFieldResponse, + WorkflowDefinitionActivateRequest, + WorkflowDefinitionCreateRequest, + WorkflowDefinitionDeleteResponse, + WorkflowDefinitionListResponse, + WorkflowDefinitionResponse, + WorkflowDefinitionRevisionListResponse, + WorkflowDefinitionRevisionResponse, + WorkflowDefinitionUpdateRequest, WorkflowDiagnosticResponse, WorkflowGraphValidationRequest, WorkflowGraphValidationResponse, @@ -18,6 +29,23 @@ from govoplan_workflow.backend.schemas import ( WorkflowNodeTypeResponse, WorkflowPortResponse, ) +from govoplan_workflow.backend.service import ( + WorkflowConflictError, + WorkflowError, + WorkflowNotFoundError, + WorkflowValidationError, + activate_definition, + archive_definition, + create_definition, + definition_response, + delete_definition, + get_definition, + get_definition_revision, + list_definition_revisions, + list_definitions, + revision_response, + update_definition, +) from govoplan_workflow.backend.validation import validate_workflow_graph @@ -33,6 +61,58 @@ def _require_any_scope(principal: ApiPrincipal, *scopes: str) -> None: ) +def _http_error(exc: WorkflowError) -> HTTPException: + if isinstance(exc, WorkflowNotFoundError): + return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) + if isinstance(exc, WorkflowConflictError): + return HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) + if isinstance(exc, WorkflowValidationError): + return HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail={ + "message": str(exc), + "diagnostics": [ + { + "severity": item.severity, + "code": item.code, + "message": item.message, + "node_id": item.node_id, + "field": item.field, + } + for item in exc.diagnostics + ], + }, + ) + return HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=str(exc), + ) + + +def _actor_id(principal: ApiPrincipal) -> str | None: + return principal.account_id or principal.membership_id or principal.identity_id + + +def _audit( + session: Session, + principal: ApiPrincipal, + *, + action: str, + definition_id: str, + details: dict[str, object], +) -> None: + audit_event( + session, + tenant_id=principal.tenant_id, + user_id=getattr(principal.user, "id", None), + api_key_id=principal.api_key_id, + action=action, + object_type="workflow_definition", + object_id=definition_id, + details=details, + ) + + @router.get("/node-types", response_model=WorkflowNodeLibraryResponse) def api_node_types( principal: ApiPrincipal = Depends(get_api_principal), @@ -123,4 +203,262 @@ def api_validate_definition( ) +@router.get("/definitions", response_model=WorkflowDefinitionListResponse) +def api_list_definitions( + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowDefinitionListResponse: + _require_any_scope(principal, DEFINITION_READ_SCOPE, ADMIN_SCOPE) + return WorkflowDefinitionListResponse( + definitions=[ + definition_response(session, definition) + for definition in list_definitions( + session, + tenant_id=principal.tenant_id, + ) + ] + ) + + +@router.post( + "/definitions", + response_model=WorkflowDefinitionResponse, + status_code=status.HTTP_201_CREATED, +) +def api_create_definition( + payload: WorkflowDefinitionCreateRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowDefinitionResponse: + _require_any_scope(principal, DEFINITION_WRITE_SCOPE, ADMIN_SCOPE) + try: + definition = create_definition( + session, + tenant_id=principal.tenant_id, + actor_id=_actor_id(principal), + payload=payload, + ) + except WorkflowError as exc: + raise _http_error(exc) from exc + _audit( + session, + principal, + action="workflow.definition.created", + definition_id=definition.id, + details={ + "key": definition.definition_key, + "revision": definition.current_revision, + }, + ) + response = definition_response(session, definition) + session.commit() + return response + + +@router.get( + "/definitions/{definition_id}", + response_model=WorkflowDefinitionResponse, +) +def api_get_definition( + definition_id: str, + revision: int | None = None, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowDefinitionResponse: + _require_any_scope(principal, DEFINITION_READ_SCOPE, ADMIN_SCOPE) + try: + definition = get_definition( + session, + tenant_id=principal.tenant_id, + definition_id=definition_id, + ) + return definition_response(session, definition, revision=revision) + except WorkflowError as exc: + raise _http_error(exc) from exc + + +@router.put( + "/definitions/{definition_id}", + response_model=WorkflowDefinitionResponse, +) +def api_update_definition( + definition_id: str, + payload: WorkflowDefinitionUpdateRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowDefinitionResponse: + _require_any_scope(principal, DEFINITION_WRITE_SCOPE, ADMIN_SCOPE) + try: + definition = update_definition( + session, + tenant_id=principal.tenant_id, + definition_id=definition_id, + actor_id=_actor_id(principal), + payload=payload, + ) + except WorkflowError as exc: + raise _http_error(exc) from exc + _audit( + session, + principal, + action="workflow.definition.updated", + definition_id=definition.id, + details={ + "revision": definition.current_revision, + "status": definition.status, + }, + ) + response = definition_response(session, definition) + session.commit() + return response + + +@router.get( + "/definitions/{definition_id}/revisions", + response_model=WorkflowDefinitionRevisionListResponse, +) +def api_list_definition_revisions( + definition_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowDefinitionRevisionListResponse: + _require_any_scope(principal, DEFINITION_READ_SCOPE, ADMIN_SCOPE) + try: + definition = get_definition( + session, + tenant_id=principal.tenant_id, + definition_id=definition_id, + ) + revisions = list_definition_revisions(session, definition=definition) + except WorkflowError as exc: + raise _http_error(exc) from exc + return WorkflowDefinitionRevisionListResponse( + revisions=[revision_response(item) for item in revisions] + ) + + +@router.get( + "/definitions/{definition_id}/revisions/{revision}", + response_model=WorkflowDefinitionRevisionResponse, +) +def api_get_definition_revision( + definition_id: str, + revision: int, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowDefinitionRevisionResponse: + _require_any_scope(principal, DEFINITION_READ_SCOPE, ADMIN_SCOPE) + try: + definition = get_definition( + session, + tenant_id=principal.tenant_id, + definition_id=definition_id, + ) + item = get_definition_revision( + session, + definition=definition, + revision=revision, + ) + except WorkflowError as exc: + raise _http_error(exc) from exc + return revision_response(item) + + +@router.post( + "/definitions/{definition_id}/activate", + response_model=WorkflowDefinitionResponse, +) +def api_activate_definition( + definition_id: str, + payload: WorkflowDefinitionActivateRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowDefinitionResponse: + _require_any_scope(principal, DEFINITION_WRITE_SCOPE, ADMIN_SCOPE) + try: + definition = activate_definition( + session, + tenant_id=principal.tenant_id, + definition_id=definition_id, + actor_id=_actor_id(principal), + revision=payload.revision, + ) + except WorkflowError as exc: + raise _http_error(exc) from exc + _audit( + session, + principal, + action="workflow.definition.activated", + definition_id=definition.id, + details={"active_revision": definition.active_revision}, + ) + response = definition_response(session, definition) + session.commit() + return response + + +@router.post( + "/definitions/{definition_id}/archive", + response_model=WorkflowDefinitionResponse, +) +def api_archive_definition( + definition_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowDefinitionResponse: + _require_any_scope(principal, DEFINITION_WRITE_SCOPE, ADMIN_SCOPE) + try: + definition = archive_definition( + session, + tenant_id=principal.tenant_id, + definition_id=definition_id, + actor_id=_actor_id(principal), + ) + except WorkflowError as exc: + raise _http_error(exc) from exc + _audit( + session, + principal, + action="workflow.definition.archived", + definition_id=definition.id, + details={"active_revision": definition.active_revision}, + ) + response = definition_response(session, definition) + session.commit() + return response + + +@router.delete( + "/definitions/{definition_id}", + response_model=WorkflowDefinitionDeleteResponse, +) +def api_delete_definition( + definition_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowDefinitionDeleteResponse: + _require_any_scope(principal, DEFINITION_WRITE_SCOPE, ADMIN_SCOPE) + try: + definition = delete_definition( + session, + tenant_id=principal.tenant_id, + definition_id=definition_id, + actor_id=_actor_id(principal), + ) + except WorkflowError as exc: + raise _http_error(exc) from exc + _audit( + session, + principal, + action="workflow.definition.deleted", + definition_id=definition.id, + details={"revision": definition.current_revision}, + ) + session.commit() + return WorkflowDefinitionDeleteResponse( + deleted=True, + definition_id=definition.id, + ) + + __all__ = ["router"] diff --git a/src/govoplan_workflow/backend/schemas.py b/src/govoplan_workflow/backend/schemas.py index 3a13468..f49f2f0 100644 --- a/src/govoplan_workflow/backend/schemas.py +++ b/src/govoplan_workflow/backend/schemas.py @@ -1,11 +1,15 @@ from __future__ import annotations import math +from datetime import datetime from typing import Any, Literal from pydantic import BaseModel, Field, field_validator +WorkflowDefinitionStatus = Literal["draft", "active", "archived"] + + class WorkflowPosition(BaseModel): x: float = 0 y: float = 0 @@ -35,6 +39,7 @@ class WorkflowEdge(BaseModel): class WorkflowGraph(BaseModel): + schema_version: Literal[1] = 1 nodes: list[WorkflowNode] = Field(default_factory=list, max_length=150) edges: list[WorkflowEdge] = Field(default_factory=list, max_length=300) @@ -91,3 +96,70 @@ class WorkflowNodeLibraryResponse(BaseModel): version: str allows_cycles: bool nodes: list[WorkflowNodeTypeResponse] + + +class WorkflowDefinitionRevisionResponse(BaseModel): + id: str + revision: int + schema_version: int + graph: WorkflowGraph + content_hash: str + library_id: str + library_version: str + created_by: str | None + created_at: datetime + + +class WorkflowDefinitionResponse(BaseModel): + id: str + tenant_id: str + key: str + name: str + description: str | None + status: WorkflowDefinitionStatus + current_revision: int + active_revision: int | None + metadata: dict[str, Any] + created_by: str | None + updated_by: str | None + created_at: datetime + updated_at: datetime + revision: WorkflowDefinitionRevisionResponse + + +class WorkflowDefinitionListResponse(BaseModel): + definitions: list[WorkflowDefinitionResponse] + + +class WorkflowDefinitionRevisionListResponse(BaseModel): + revisions: list[WorkflowDefinitionRevisionResponse] + + +class WorkflowDefinitionCreateRequest(BaseModel): + key: str | None = Field( + default=None, + min_length=1, + max_length=120, + pattern=r"^[A-Za-z0-9][A-Za-z0-9_-]*$", + ) + name: str = Field(min_length=1, max_length=300) + description: str | None = Field(default=None, max_length=4_000) + graph: WorkflowGraph + metadata: dict[str, Any] = Field(default_factory=dict) + + +class WorkflowDefinitionUpdateRequest(BaseModel): + name: str = Field(min_length=1, max_length=300) + description: str | None = Field(default=None, max_length=4_000) + graph: WorkflowGraph + metadata: dict[str, Any] = Field(default_factory=dict) + expected_revision: int = Field(ge=1) + + +class WorkflowDefinitionActivateRequest(BaseModel): + revision: int | None = Field(default=None, ge=1) + + +class WorkflowDefinitionDeleteResponse(BaseModel): + deleted: bool + definition_id: str diff --git a/src/govoplan_workflow/backend/service.py b/src/govoplan_workflow/backend/service.py new file mode 100644 index 0000000..4702be0 --- /dev/null +++ b/src/govoplan_workflow/backend/service.py @@ -0,0 +1,399 @@ +from __future__ import annotations + +import hashlib +import json +import re +import unicodedata + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from govoplan_core.db.base import utcnow +from govoplan_workflow.backend.db.models import ( + WorkflowDefinition, + WorkflowDefinitionRevision, +) +from govoplan_workflow.backend.node_library import WORKFLOW_GRAPH_LIBRARY +from govoplan_workflow.backend.schemas import ( + WorkflowDefinitionCreateRequest, + WorkflowDefinitionResponse, + WorkflowDefinitionRevisionResponse, + WorkflowDefinitionUpdateRequest, + WorkflowGraph, +) +from govoplan_workflow.backend.validation import validate_workflow_graph + + +class WorkflowError(RuntimeError): + pass + + +class WorkflowNotFoundError(WorkflowError): + pass + + +class WorkflowConflictError(WorkflowError): + pass + + +class WorkflowValidationError(WorkflowError): + def __init__(self, diagnostics: tuple[object, ...]) -> None: + first = diagnostics[0] if diagnostics else None + super().__init__( + str(getattr(first, "message", "Workflow definition validation failed.")) + ) + self.diagnostics = diagnostics + + +def list_definitions( + session: Session, + *, + tenant_id: str, +) -> list[WorkflowDefinition]: + return list( + session.scalars( + select(WorkflowDefinition) + .where( + WorkflowDefinition.tenant_id == tenant_id, + WorkflowDefinition.deleted_at.is_(None), + ) + .order_by( + WorkflowDefinition.updated_at.desc(), + WorkflowDefinition.name.asc(), + ) + ) + ) + + +def get_definition( + session: Session, + *, + tenant_id: str, + definition_id: str, +) -> WorkflowDefinition: + definition = session.scalar( + select(WorkflowDefinition).where( + WorkflowDefinition.id == definition_id, + WorkflowDefinition.tenant_id == tenant_id, + WorkflowDefinition.deleted_at.is_(None), + ) + ) + if definition is None: + raise WorkflowNotFoundError("Workflow definition not found.") + return definition + + +def get_definition_revision( + session: Session, + *, + definition: WorkflowDefinition, + revision: int | None = None, +) -> WorkflowDefinitionRevision: + revision_number = revision or definition.current_revision + item = session.scalar( + select(WorkflowDefinitionRevision).where( + WorkflowDefinitionRevision.definition_id == definition.id, + WorkflowDefinitionRevision.tenant_id == definition.tenant_id, + WorkflowDefinitionRevision.revision == revision_number, + ) + ) + if item is None: + raise WorkflowNotFoundError("Workflow definition revision not found.") + return item + + +def list_definition_revisions( + session: Session, + *, + definition: WorkflowDefinition, +) -> list[WorkflowDefinitionRevision]: + return list( + session.scalars( + select(WorkflowDefinitionRevision) + .where( + WorkflowDefinitionRevision.definition_id == definition.id, + WorkflowDefinitionRevision.tenant_id == definition.tenant_id, + ) + .order_by(WorkflowDefinitionRevision.revision.desc()) + ) + ) + + +def create_definition( + session: Session, + *, + tenant_id: str, + actor_id: str | None, + payload: WorkflowDefinitionCreateRequest, +) -> WorkflowDefinition: + graph = _validated_graph(payload.graph) + definition = WorkflowDefinition( + tenant_id=tenant_id, + definition_key=_available_key( + session, + tenant_id=tenant_id, + requested=payload.key, + name=payload.name, + ), + name=payload.name.strip(), + description=_clean_optional(payload.description), + status="draft", + current_revision=1, + active_revision=None, + metadata_=dict(payload.metadata), + created_by=actor_id, + updated_by=actor_id, + ) + definition.revisions.append( + _new_revision( + tenant_id=tenant_id, + revision=1, + graph=graph, + actor_id=actor_id, + ) + ) + session.add(definition) + session.flush() + return definition + + +def update_definition( + session: Session, + *, + tenant_id: str, + definition_id: str, + actor_id: str | None, + payload: WorkflowDefinitionUpdateRequest, +) -> WorkflowDefinition: + definition = get_definition( + session, + tenant_id=tenant_id, + definition_id=definition_id, + ) + if payload.expected_revision != definition.current_revision: + raise WorkflowConflictError( + "Workflow definition changed on the server; " + f"expected revision {payload.expected_revision}, " + f"current revision is {definition.current_revision}." + ) + graph = _validated_graph(payload.graph) + current = get_definition_revision(session, definition=definition) + graph_hash = _content_hash(graph) + definition.name = payload.name.strip() + definition.description = _clean_optional(payload.description) + definition.metadata_ = dict(payload.metadata) + definition.updated_by = actor_id + if current.content_hash != graph_hash: + definition.current_revision += 1 + definition.revisions.append( + _new_revision( + tenant_id=tenant_id, + revision=definition.current_revision, + graph=graph, + actor_id=actor_id, + ) + ) + if definition.status == "active": + definition.status = "draft" + session.flush() + return definition + + +def activate_definition( + session: Session, + *, + tenant_id: str, + definition_id: str, + actor_id: str | None, + revision: int | None = None, +) -> WorkflowDefinition: + definition = get_definition( + session, + tenant_id=tenant_id, + definition_id=definition_id, + ) + selected = get_definition_revision( + session, + definition=definition, + revision=revision, + ) + _validated_graph(WorkflowGraph.model_validate(selected.graph)) + definition.active_revision = selected.revision + definition.status = "active" + definition.updated_by = actor_id + session.flush() + return definition + + +def archive_definition( + session: Session, + *, + tenant_id: str, + definition_id: str, + actor_id: str | None, +) -> WorkflowDefinition: + definition = get_definition( + session, + tenant_id=tenant_id, + definition_id=definition_id, + ) + definition.status = "archived" + definition.updated_by = actor_id + session.flush() + return definition + + +def delete_definition( + session: Session, + *, + tenant_id: str, + definition_id: str, + actor_id: str | None, +) -> WorkflowDefinition: + definition = get_definition( + session, + tenant_id=tenant_id, + definition_id=definition_id, + ) + definition.deleted_at = utcnow() + definition.updated_by = actor_id + session.flush() + return definition + + +def definition_response( + session: Session, + definition: WorkflowDefinition, + *, + revision: int | None = None, +) -> WorkflowDefinitionResponse: + selected = get_definition_revision( + session, + definition=definition, + revision=revision, + ) + return WorkflowDefinitionResponse( + id=definition.id, + tenant_id=definition.tenant_id, + key=definition.definition_key, + name=definition.name, + description=definition.description, + status=definition.status, + current_revision=definition.current_revision, + active_revision=definition.active_revision, + metadata=dict(definition.metadata_), + created_by=definition.created_by, + updated_by=definition.updated_by, + created_at=definition.created_at, + updated_at=definition.updated_at, + revision=revision_response(selected), + ) + + +def revision_response( + revision: WorkflowDefinitionRevision, +) -> WorkflowDefinitionRevisionResponse: + return WorkflowDefinitionRevisionResponse( + id=revision.id, + revision=revision.revision, + schema_version=revision.schema_version, + graph=WorkflowGraph.model_validate(revision.graph), + content_hash=revision.content_hash, + library_id=revision.library_id, + library_version=revision.library_version, + created_by=revision.created_by, + created_at=revision.created_at, + ) + + +def _new_revision( + *, + tenant_id: str, + revision: int, + graph: WorkflowGraph, + actor_id: str | None, +) -> WorkflowDefinitionRevision: + return WorkflowDefinitionRevision( + tenant_id=tenant_id, + revision=revision, + schema_version=graph.schema_version, + graph=_canonical_graph(graph), + content_hash=_content_hash(graph), + library_id=WORKFLOW_GRAPH_LIBRARY.id, + library_version=WORKFLOW_GRAPH_LIBRARY.version, + created_by=actor_id, + ) + + +def _validated_graph(graph: WorkflowGraph) -> WorkflowGraph: + diagnostics = validate_workflow_graph(graph) + if any(item.severity == "error" for item in diagnostics): + raise WorkflowValidationError(diagnostics) + return graph + + +def _canonical_graph(graph: WorkflowGraph) -> dict[str, object]: + return graph.model_dump(mode="json") + + +def _content_hash(graph: WorkflowGraph) -> str: + encoded = json.dumps( + _canonical_graph(graph), + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def _available_key( + session: Session, + *, + tenant_id: str, + requested: str | None, + name: str, +) -> str: + base = _slug(requested or name) + candidate = base + suffix = 2 + while session.scalar( + select(WorkflowDefinition.id).where( + WorkflowDefinition.tenant_id == tenant_id, + WorkflowDefinition.definition_key == candidate, + ) + ): + candidate = f"{base[: max(1, 120 - len(str(suffix)) - 1)]}-{suffix}" + suffix += 1 + return candidate + + +def _slug(value: str) -> str: + normalized = unicodedata.normalize("NFKD", value) + ascii_value = normalized.encode("ascii", "ignore").decode("ascii").lower() + cleaned = re.sub(r"[^a-z0-9]+", "-", ascii_value).strip("-") + return (cleaned or "workflow")[:120] + + +def _clean_optional(value: str | None) -> str | None: + if value is None: + return None + cleaned = value.strip() + return cleaned or None + + +__all__ = [ + "WorkflowConflictError", + "WorkflowError", + "WorkflowNotFoundError", + "WorkflowValidationError", + "activate_definition", + "archive_definition", + "create_definition", + "definition_response", + "delete_definition", + "get_definition", + "get_definition_revision", + "list_definition_revisions", + "list_definitions", + "revision_response", + "update_definition", +] diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 0b69042..b31c4a8 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -26,6 +26,17 @@ class WorkflowManifestTests(unittest.TestCase): DEFINITION_WRITE_SCOPE, {item.scope for item in manifest.permissions}, ) + self.assertEqual( + "@govoplan/workflow-webui", + manifest.frontend.package_name if manifest.frontend else None, + ) + self.assertIsNotNone(manifest.migration_spec) + requirement = next( + item + for item in manifest.requires_interfaces + if item.name == "dataflow.run_lifecycle" + ) + self.assertTrue(requirement.optional) if __name__ == "__main__": diff --git a/tests/test_migrations.py b/tests/test_migrations.py new file mode 100644 index 0000000..024f408 --- /dev/null +++ b/tests/test_migrations.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from alembic.runtime.migration import MigrationContext +from sqlalchemy import create_engine, inspect + +from govoplan_core.db.migrations import migrate_database +from govoplan_workflow.backend.manifest import get_manifest + + +class WorkflowMigrationTests(unittest.TestCase): + def test_migration_creates_definition_tables_and_head(self) -> None: + with tempfile.TemporaryDirectory( + prefix="govoplan-workflow-migration-" + ) as directory: + url = f"sqlite:///{Path(directory) / 'workflow.db'}" + migrate_database( + database_url=url, + enabled_modules=("workflow",), + manifest_factories=(get_manifest,), + ) + engine = create_engine(url) + try: + with engine.connect() as connection: + self.assertIn( + "a7c4e2f9b1d3", + set(MigrationContext.configure(connection).get_current_heads()), + ) + self.assertEqual( + { + "workflow_definition_revisions", + "workflow_definitions", + }, + { + name + for name in inspect(connection).get_table_names() + if name.startswith("workflow_") + }, + ) + finally: + engine.dispose() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_service.py b/tests/test_service.py new file mode 100644 index 0000000..07a1457 --- /dev/null +++ b/tests/test_service.py @@ -0,0 +1,225 @@ +from __future__ import annotations + +import unittest + +from sqlalchemy import create_engine, select +from sqlalchemy.orm import Session, sessionmaker + +from govoplan_core.db.base import Base +from govoplan_workflow.backend.db.models import ( + WorkflowDefinition, + WorkflowDefinitionRevision, +) +from govoplan_workflow.backend.schemas import ( + WorkflowDefinitionCreateRequest, + WorkflowDefinitionUpdateRequest, + WorkflowEdge, + WorkflowGraph, + WorkflowNode, + WorkflowPosition, +) +from govoplan_workflow.backend.service import ( + WorkflowConflictError, + WorkflowNotFoundError, + activate_definition, + create_definition, + delete_definition, + get_definition, + list_definition_revisions, + list_definitions, + update_definition, +) + + +def sample_graph(*, title: str = "Review request") -> WorkflowGraph: + return WorkflowGraph( + nodes=[ + WorkflowNode( + id="start", + type="workflow.start.manual", + label="Start", + position=WorkflowPosition(x=40, y=100), + config={"input_schema_ref": ""}, + ), + WorkflowNode( + id="activity", + type="workflow.activity", + label="Review", + position=WorkflowPosition(x=280, y=100), + config={ + "title": title, + "instructions": "", + "assignee": "", + "due_after": "", + }, + ), + WorkflowNode( + id="complete", + type="workflow.end.completed", + label="Completed", + position=WorkflowPosition(x=520, y=100), + config={"output_mapping": {}}, + ), + ], + edges=[ + WorkflowEdge(id="start-activity", source="start", target="activity"), + WorkflowEdge(id="activity-complete", source="activity", target="complete"), + ], + ) + + +class WorkflowServiceTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all( + self.engine, + tables=[ + WorkflowDefinition.__table__, + WorkflowDefinitionRevision.__table__, + ], + ) + self.Session = sessionmaker(bind=self.engine) + self.session: Session = self.Session() + + def tearDown(self) -> None: + self.session.close() + Base.metadata.drop_all( + self.engine, + tables=[ + WorkflowDefinitionRevision.__table__, + WorkflowDefinition.__table__, + ], + ) + self.engine.dispose() + + def _create(self, *, tenant_id: str = "tenant-1") -> WorkflowDefinition: + definition = create_definition( + self.session, + tenant_id=tenant_id, + actor_id="user-1", + payload=WorkflowDefinitionCreateRequest( + name="Monthly case handling", + graph=sample_graph(), + ), + ) + self.session.commit() + return definition + + def test_create_update_and_activate_pin_immutable_revisions(self) -> None: + definition = self._create() + updated = update_definition( + self.session, + tenant_id="tenant-1", + definition_id=definition.id, + actor_id="user-2", + payload=WorkflowDefinitionUpdateRequest( + name="Monthly case handling", + graph=sample_graph(title="Review corrected request"), + expected_revision=1, + ), + ) + activate_definition( + self.session, + tenant_id="tenant-1", + definition_id=definition.id, + actor_id="user-2", + revision=1, + ) + self.session.commit() + + revisions = list_definition_revisions( + self.session, + definition=updated, + ) + self.assertEqual(2, updated.current_revision) + self.assertEqual(1, updated.active_revision) + self.assertEqual("active", updated.status) + self.assertEqual([2, 1], [item.revision for item in revisions]) + self.assertNotEqual(revisions[0].content_hash, revisions[1].content_hash) + historical = next(item for item in revisions if item.revision == 1) + self.assertEqual( + "Review request", + historical.graph["nodes"][1]["config"]["title"], + ) + + def test_metadata_update_does_not_create_graph_revision(self) -> None: + definition = self._create() + updated = update_definition( + self.session, + tenant_id="tenant-1", + definition_id=definition.id, + actor_id="user-2", + payload=WorkflowDefinitionUpdateRequest( + name="Renamed workflow", + description="Updated metadata only", + graph=sample_graph(), + metadata={"owner": "finance"}, + expected_revision=1, + ), + ) + self.session.commit() + + self.assertEqual(1, updated.current_revision) + self.assertEqual( + 1, + len( + list( + self.session.scalars( + select(WorkflowDefinitionRevision).where( + WorkflowDefinitionRevision.definition_id + == definition.id + ) + ) + ) + ), + ) + + def test_stale_update_and_cross_tenant_access_are_rejected(self) -> None: + definition = self._create() + with self.assertRaises(WorkflowConflictError): + update_definition( + self.session, + tenant_id="tenant-1", + definition_id=definition.id, + actor_id="user-2", + payload=WorkflowDefinitionUpdateRequest( + name="Stale", + graph=sample_graph(), + expected_revision=2, + ), + ) + with self.assertRaises(WorkflowNotFoundError): + get_definition( + self.session, + tenant_id="tenant-2", + definition_id=definition.id, + ) + + def test_soft_delete_preserves_revisions_and_hides_definition(self) -> None: + definition = self._create() + delete_definition( + self.session, + tenant_id="tenant-1", + definition_id=definition.id, + actor_id="user-2", + ) + self.session.commit() + + self.assertEqual([], list_definitions(self.session, tenant_id="tenant-1")) + self.assertEqual( + 1, + len( + list( + self.session.scalars( + select(WorkflowDefinitionRevision).where( + WorkflowDefinitionRevision.definition_id + == definition.id + ) + ) + ) + ), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/package.json b/webui/package.json new file mode 100644 index 0000000..852d60e --- /dev/null +++ b/webui/package.json @@ -0,0 +1,32 @@ +{ + "name": "@govoplan/workflow-webui", + "version": "0.1.14", + "private": true, + "type": "module", + "main": "src/index.ts", + "module": "src/index.ts", + "types": "src/index.ts", + "exports": { + ".": { + "types": "./src/index.ts", + "import": "./src/index.ts" + } + }, + "scripts": { + "typecheck": "tsc --noEmit" + }, + "peerDependencies": { + "@govoplan/core-webui": "^0.1.14", + "@xyflow/react": "^12.11.2", + "lucide-react": "^1.23.0", + "react": "^19.0.0", + "react-dom": "^19.0.0", + "react-router-dom": "^7.1.1", + "typescript": "^5.7.2" + }, + "peerDependenciesMeta": { + "@govoplan/core-webui": { + "optional": true + } + } +} diff --git a/webui/src/api/workflow.ts b/webui/src/api/workflow.ts new file mode 100644 index 0000000..ff574b3 --- /dev/null +++ b/webui/src/api/workflow.ts @@ -0,0 +1,158 @@ +import { apiFetch, type ApiSettings } from "@govoplan/core-webui"; +import type { + DefinitionGraph, + DefinitionGraphNode, + DefinitionGraphNodeType +} from "@govoplan/core-webui/definition-graph"; + +export type WorkflowStatus = "draft" | "active" | "archived"; +export type WorkflowGraphNode = DefinitionGraphNode; +export type WorkflowGraph = DefinitionGraph & { + schema_version: 1; + nodes: WorkflowGraphNode[]; +}; + +export type WorkflowDiagnostic = { + severity: "error" | "warning"; + code: string; + message: string; + node_id?: string | null; + field?: string | null; +}; + +export type WorkflowNodeType = DefinitionGraphNodeType; + +export type WorkflowRevision = { + id: string; + revision: number; + schema_version: number; + graph: WorkflowGraph; + content_hash: string; + library_id: string; + library_version: string; + created_by?: string | null; + created_at: string; +}; + +export type WorkflowDefinition = { + id: string; + tenant_id: string; + key: string; + name: string; + description?: string | null; + status: WorkflowStatus; + current_revision: number; + active_revision?: number | null; + metadata: Record; + created_by?: string | null; + updated_by?: string | null; + created_at: string; + updated_at: string; + revision: WorkflowRevision; +}; + +export type WorkflowDefinitionPayload = { + name: string; + description?: string | null; + graph: WorkflowGraph; + metadata: Record; +}; + +export async function listWorkflowNodeTypes( + settings: ApiSettings +): Promise<{ id: string; version: string; allows_cycles: boolean; nodes: WorkflowNodeType[] }> { + return apiFetch(settings, "/api/v1/workflow/node-types"); +} + +export async function listWorkflowDefinitions( + settings: ApiSettings +): Promise { + const response = await apiFetch<{ definitions: WorkflowDefinition[] }>( + settings, + "/api/v1/workflow/definitions" + ); + return response.definitions; +} + +export function createWorkflowDefinition( + settings: ApiSettings, + payload: WorkflowDefinitionPayload +): Promise { + return apiFetch(settings, "/api/v1/workflow/definitions", { + method: "POST", + body: JSON.stringify(payload) + }); +} + +export function updateWorkflowDefinition( + settings: ApiSettings, + definitionId: string, + payload: WorkflowDefinitionPayload & { expected_revision: number } +): Promise { + return apiFetch( + settings, + `/api/v1/workflow/definitions/${encodeURIComponent(definitionId)}`, + { + method: "PUT", + body: JSON.stringify(payload) + } + ); +} + +export async function listWorkflowRevisions( + settings: ApiSettings, + definitionId: string +): Promise { + const response = await apiFetch<{ revisions: WorkflowRevision[] }>( + settings, + `/api/v1/workflow/definitions/${encodeURIComponent(definitionId)}/revisions` + ); + return response.revisions; +} + +export function activateWorkflowDefinition( + settings: ApiSettings, + definitionId: string, + revision?: number +): Promise { + return apiFetch( + settings, + `/api/v1/workflow/definitions/${encodeURIComponent(definitionId)}/activate`, + { + method: "POST", + body: JSON.stringify({ revision }) + } + ); +} + +export function archiveWorkflowDefinition( + settings: ApiSettings, + definitionId: string +): Promise { + return apiFetch( + settings, + `/api/v1/workflow/definitions/${encodeURIComponent(definitionId)}/archive`, + { method: "POST" } + ); +} + +export function deleteWorkflowDefinition( + settings: ApiSettings, + definitionId: string +): Promise<{ deleted: boolean; definition_id: string }> { + return apiFetch( + settings, + `/api/v1/workflow/definitions/${encodeURIComponent(definitionId)}`, + { method: "DELETE" } + ); +} + +export function validateWorkflowDefinition( + settings: ApiSettings, + graph: WorkflowGraph +): Promise<{ valid: boolean; diagnostics: WorkflowDiagnostic[] }> { + return apiFetch(settings, "/api/v1/workflow/definitions/validate", { + method: "POST", + body: JSON.stringify({ graph }) + }); +} diff --git a/webui/src/features/workflow/WorkflowCanvas.tsx b/webui/src/features/workflow/WorkflowCanvas.tsx new file mode 100644 index 0000000..5a26088 --- /dev/null +++ b/webui/src/features/workflow/WorkflowCanvas.tsx @@ -0,0 +1,229 @@ +import { useMemo, useState, type DragEvent } from "react"; +import { + addEdge, + applyEdgeChanges, + applyNodeChanges, + Background, + BackgroundVariant, + ConnectionLineType, + Controls, + MiniMap, + ReactFlow, + type Connection, + type Edge, + type ReactFlowInstance +} from "@xyflow/react"; +import { definitionConnectionError } from "@govoplan/core-webui/definition-graph"; +import type { + WorkflowDiagnostic, + WorkflowGraph, + WorkflowGraphNode, + WorkflowNodeType +} from "../../api/workflow"; +import { newWorkflowNode } from "./model"; +import WorkflowNode, { type WorkflowFlowNode } from "./WorkflowNode"; + +const nodeTypes = { workflow: WorkflowNode }; + +export default function WorkflowCanvas({ + graph, + diagnostics, + nodeLibrary, + selectedNodeId, + readOnly, + allowsCycles, + onGraphChange, + onSelectNode +}: { + graph: WorkflowGraph; + diagnostics: WorkflowDiagnostic[]; + nodeLibrary: WorkflowNodeType[]; + selectedNodeId: string | null; + readOnly: boolean; + allowsCycles: boolean; + onGraphChange: (graph: WorkflowGraph) => void; + onSelectNode: (nodeId: string | null) => void; +}) { + const [instance, setInstance] = useState< + ReactFlowInstance | null + >(null); + const definitions = useMemo( + () => new Map(nodeLibrary.map((item) => [item.type, item])), + [nodeLibrary] + ); + const errorNodeIds = useMemo( + () => new Set( + diagnostics + .filter((item) => item.severity === "error" && item.node_id) + .map((item) => item.node_id as string) + ), + [diagnostics] + ); + const nodes = useMemo( + () => graph.nodes.flatMap((node) => { + const definition = definitions.get(node.type); + if (!definition) return []; + return [{ + id: node.id, + type: "workflow" as const, + position: node.position, + selected: node.id === selectedNodeId, + data: { + label: node.label, + workflowType: node.type, + definition, + hasError: errorNodeIds.has(node.id) + } + }]; + }), + [definitions, errorNodeIds, graph.nodes, selectedNodeId] + ); + const edges = useMemo( + () => graph.edges.map((edge) => ({ + id: edge.id, + source: edge.source, + target: edge.target, + sourceHandle: edge.source_port ?? "output", + targetHandle: edge.target_port ?? "input", + type: "smoothstep", + className: "workflow-edge" + })), + [graph.edges] + ); + + const updateNodes = (nextNodes: WorkflowFlowNode[]) => { + const ids = new Set(nextNodes.map((node) => node.id)); + onGraphChange({ + ...graph, + nodes: nextNodes.map((flowNode) => { + const current = graph.nodes.find((node) => node.id === flowNode.id); + if (!current) throw new Error(`Unknown Workflow node: ${flowNode.id}`); + return { ...current, position: flowNode.position }; + }), + edges: graph.edges.filter( + (edge) => ids.has(edge.source) && ids.has(edge.target) + ) + }); + }; + + const updateEdges = (nextEdges: Edge[]) => { + onGraphChange({ + ...graph, + edges: nextEdges.map((edge) => ({ + id: edge.id, + source: edge.source, + target: edge.target, + source_port: edge.sourceHandle ?? "output", + target_port: edge.targetHandle ?? "input" + })) + }); + }; + + const isValidConnection = (connection: Connection | Edge): boolean => { + if (readOnly || !connection.source || !connection.target) return false; + return definitionConnectionError( + graph, + nodeLibrary, + { + source: connection.source, + target: connection.target, + sourcePort: connection.sourceHandle, + targetPort: connection.targetHandle + }, + { allowCycles: allowsCycles } + ) === null; + }; + + const onDrop = (event: DragEvent) => { + event.preventDefault(); + if (readOnly || !instance) return; + const type = event.dataTransfer.getData( + "application/x-govoplan-workflow-node" + ); + if (!type) return; + const node = newWorkflowNode( + type, + instance.screenToFlowPosition({ x: event.clientX, y: event.clientY }), + nodeLibrary + ); + onGraphChange({ ...graph, nodes: [...graph.nodes, node] }); + onSelectNode(node.id); + }; + + return ( +
{ + if ( + event.dataTransfer.types.includes( + "application/x-govoplan-workflow-node" + ) + ) { + event.preventDefault(); + event.dataTransfer.dropEffect = "copy"; + } + }} + onDrop={onDrop} + > + + nodes={nodes} + edges={edges} + nodeTypes={nodeTypes} + onInit={setInstance} + onNodesChange={(changes) => { + if (!readOnly) updateNodes(applyNodeChanges(changes, nodes)); + }} + onEdgesChange={(changes) => { + if (!readOnly) updateEdges(applyEdgeChanges(changes, edges)); + }} + onConnect={(connection) => { + if (!isValidConnection(connection)) return; + updateEdges(addEdge({ + ...connection, + id: `edge-${crypto.randomUUID()}`, + type: "smoothstep" + }, edges)); + }} + isValidConnection={isValidConnection} + onNodeClick={(_event, node) => onSelectNode(node.id)} + onPaneClick={() => onSelectNode(null)} + nodesDraggable={!readOnly} + nodesConnectable={!readOnly} + deleteKeyCode={readOnly ? null : ["Backspace", "Delete"]} + connectionLineType={ConnectionLineType.SmoothStep} + connectionLineStyle={{ stroke: "var(--accent)", strokeWidth: 3 }} + connectionRadius={32} + fitView + fitViewOptions={{ padding: 0.22, maxZoom: 1.25 }} + minZoom={0.25} + maxZoom={1.8} + > + + + node.data.hasError ? "var(--danger)" : "var(--accent)" + } + /> + + + {!graph.nodes.length ? ( +
Drop a start node here
+ ) : null} +
+ ); +} + +export function updateWorkflowGraphNode( + graph: WorkflowGraph, + updatedNode: WorkflowGraphNode +): WorkflowGraph { + return { + ...graph, + nodes: graph.nodes.map((node) => + node.id === updatedNode.id ? updatedNode : node + ) + }; +} diff --git a/webui/src/features/workflow/WorkflowInspector.tsx b/webui/src/features/workflow/WorkflowInspector.tsx new file mode 100644 index 0000000..55420d6 --- /dev/null +++ b/webui/src/features/workflow/WorkflowInspector.tsx @@ -0,0 +1,185 @@ +import { useEffect, useState } from "react"; +import { Trash2 } from "lucide-react"; +import { + Button, + DismissibleAlert, + FormField +} from "@govoplan/core-webui"; +import type { + WorkflowGraphNode, + WorkflowNodeType +} from "../../api/workflow"; + +export default function WorkflowInspector({ + node, + nodeLibrary, + readOnly, + onChange, + onDelete +}: { + node: WorkflowGraphNode | null; + nodeLibrary: WorkflowNodeType[]; + readOnly: boolean; + onChange: (node: WorkflowGraphNode) => void; + onDelete: (nodeId: string) => void; +}) { + const [jsonDrafts, setJsonDrafts] = useState>({}); + const [error, setError] = useState(""); + + useEffect(() => { + if (!node) { + setJsonDrafts({}); + setError(""); + return; + } + const definition = nodeLibrary.find((item) => item.type === node.type); + setJsonDrafts(Object.fromEntries( + (definition?.config_fields ?? []) + .filter((field) => field.kind === "mapping") + .map((field) => [ + field.id, + JSON.stringify(node.config[field.id] ?? {}, null, 2) + ]) + )); + setError(""); + }, [node?.id, nodeLibrary]); + + if (!node) { + return ( + + ); + } + + const definition = nodeLibrary.find((item) => item.type === node.type); + const updateConfig = (field: string, value: unknown) => { + onChange({ + ...node, + config: { ...node.config, [field]: value } + }); + }; + + return ( +