govoplan-dataflow/src/govoplan_dataflow/backend/manifest.py

from __future__ import annotations

from govoplan_core.core.modules import with_documentation_structured_translations
from govoplan_dataflow.backend.german_structured_documentation import GERMAN_STRUCTURED_TRANSLATIONS

from pathlib import Path

from govoplan_core.core.module_guards import (
    drop_table_retirement_provider,
    persistent_table_uninstall_guard,
)
from govoplan_core.core.access import (
    CAPABILITY_ACCESS_DIRECTORY,
    CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
)
from govoplan_core.core.dataflows import (
    CAPABILITY_DATAFLOW_DATASET_OUTPUT,
    CAPABILITY_DATAFLOW_RUN_LIFECYCLE,
    CAPABILITY_DATAFLOW_RUN_WORKER,
    CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
)
from govoplan_core.core.modules import (
    CapabilityDocumentation,
    DocumentationCondition,
    DocumentationTopic,
    FrontendModule,
    FrontendRoute,
    MigrationSpec,
    ModuleInterfaceProvider,
    ModuleInterfaceRequirement,
    ModuleContext,
    ModuleManifest,
    NavItem,
    PermissionDefinition,
    ProductAreaContribution,
    RoleTemplate,
)
from govoplan_core.core.provider_governance import declared_module_architecture
from govoplan_core.core.datasources import (
    CAPABILITY_DATASOURCE_CATALOGUE,
    CAPABILITY_DATASOURCE_LIFECYCLE,
    CAPABILITY_DATASOURCE_PUBLICATION,
)
from govoplan_core.core.policy import (
    CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
)
from govoplan_core.core.notifications import (
    CAPABILITY_NOTIFICATIONS_DISPATCH,
)
from govoplan_core.core.references import CAPABILITY_ACCESS_REFERENCE_OPTIONS
from govoplan_core.core.search import SearchSourceProviderRegistration
from govoplan_core.core.views import ViewSurface
from govoplan_core.db.base import Base
from govoplan_dataflow.backend.db import models as dataflow_models
from govoplan_dataflow.backend.dsar_provider import (
    DATAFLOW_DSAR_CAPABILITY,
    DataflowDsarProvider,
)
from govoplan_dataflow.backend.german_documentation import (
    localize_documentation_topics,
)


MODULE_ID = "dataflow"
MODULE_NAME = "Dataflow"
MODULE_VERSION = "0.1.25"

READ_SCOPE = "dataflow:pipeline:read"
WRITE_SCOPE = "dataflow:pipeline:write"
RUN_SCOPE = "dataflow:pipeline:run"
ADMIN_SCOPE = "dataflow:pipeline:admin"
TRIGGER_READ_SCOPE = "dataflow:trigger:read"
TRIGGER_WRITE_SCOPE = "dataflow:trigger:write"
TRIGGER_DISPATCH_SCOPE = "dataflow:trigger:dispatch"


def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
    module_id, resource, action = scope.split(":", 2)
    return PermissionDefinition(
        scope=scope,
        label=label,
        description=description,
        category="Dataflow",
        level="tenant",
        module_id=module_id,
        resource=resource,
        action=action,
    )


PERMISSIONS = (
    _permission(
        READ_SCOPE,
        "View data pipelines",
        "Read pipeline definitions, revisions, validation diagnostics, and run summaries.",
    ),
    _permission(
        WRITE_SCOPE,
        "Manage data pipelines",
        "Create, edit, version, archive, and remove data pipeline definitions.",
    ),
    _permission(
        RUN_SCOPE,
        "Run data pipelines",
        "Validate and execute bounded previews or approved pipeline runs.",
    ),
    _permission(
        ADMIN_SCOPE,
        "Administer Dataflow",
        "Manage every tenant pipeline and future execution, retention, and publication policies.",
    ),
    _permission(
        TRIGGER_READ_SCOPE,
        "View Dataflow triggers",
        "Inspect schedule and event trigger configuration, status, provenance, and deliveries.",
    ),
    _permission(
        TRIGGER_WRITE_SCOPE,
        "Manage Dataflow triggers",
        "Create, edit, enable, disable, and remove governed Dataflow triggers.",
    ),
    _permission(
        TRIGGER_DISPATCH_SCOPE,
        "Dispatch Dataflow triggers",
        "Ingest trusted events and dispatch queued automated Dataflow runs.",
    ),
)

ROLE_TEMPLATES = (
    RoleTemplate(
        slug="dataflow_designer",
        name="Dataflow designer",
        description="Design, validate, and preview governed data pipelines.",
        permissions=(
            READ_SCOPE,
            WRITE_SCOPE,
            RUN_SCOPE,
            TRIGGER_READ_SCOPE,
            TRIGGER_WRITE_SCOPE,
        ),
    ),
    RoleTemplate(
        slug="dataflow_operator",
        name="Dataflow operator",
        description="Inspect and run approved data pipelines without changing their definitions.",
        permissions=(READ_SCOPE, RUN_SCOPE, TRIGGER_READ_SCOPE),
    ),
    RoleTemplate(
        slug="dataflow_viewer",
        name="Dataflow viewer",
        description="Inspect pipeline definitions, revisions, diagnostics, and run summaries.",
        permissions=(READ_SCOPE, TRIGGER_READ_SCOPE),
    ),
)

DOCUMENTATION = localize_documentation_topics((
    DocumentationTopic(
        id="dataflow.csv-source-fidelity",
        title="Import CSV without silently changing its values",
        summary="Choose text preservation or explicit legacy inference before creating a durable datasource.",
        body=(
            "The CSV import dialog defaults to Preserve text (no automatic conversion). Field whitespace, decimal digits, large identifiers, boolean-looking text and explicit empty records remain strings. "
            "Choose Infer types (legacy) only when you want the existing numeric/boolean conversion and empty-row rules. JSON imports are unchanged. Existing API clients omitting csv_value_mode retain legacy_typed behavior. "
            "The imported datasource is a deliberate durable upload, not a retained preview. Datasources owns its original UTF-8 CSV text, parsed rows, approval evidence, immutable materialization and retention; Core verifies that original input and parsed scalar types/values agree. "
            "Both original and parsed content are bounded to 5 MB and the table to 10,000 rows. Original-source export is available through the Datasources administrator API only with unrestricted current and historical visibility, never through catalogue metadata. "
            "Missing older originals cannot be reconstructed. Exact text mode can change the inferred schema to strings; review downstream numeric comparisons and conversions before adopting the new source."
        ),
        layer="always", documentation_types=("user", "admin"), audience=("user", "module_admin", "operator"), order=8,
        translations={"de": {
            "title": "CSV importieren, ohne Werte unbemerkt zu ändern",
            "summary": "Vor dem dauerhaften Import zwischen Texterhalt und ausdrücklicher bisheriger Typableitung wählen.",
            "body": (
                "Der CSV-Import wählt standardmäßig Text erhalten (keine automatische Umwandlung). Leerzeichen in Werten, Dezimalstellen, große Kennungen, boolesch wirkender Text und ausdrücklich leere Datensätze bleiben Zeichenketten. "
                "Wählen Sie Typen ableiten (bisheriges Verhalten), wenn Sie die bisherige Zahlen-/Wahrheitswertumwandlung und Behandlung leerer Zeilen benötigen. JSON-Importe bleiben unverändert; API-Aufrufe ohne csv_value_mode behalten legacy_typed. "
                "Der Import ist eine bewusst dauerhafte Datenquelle und keine gespeicherte Vorschau. Datasources verantwortet Originaltext als UTF-8, verarbeitete Zeilen, Freigabenachweis, unveränderliche Materialisierung und Aufbewahrung. Core prüft die Übereinstimmung von Quelltext sowie genauen Typen und Werten. "
                "Original und verarbeiteter Inhalt sind jeweils auf 5 MB, die Tabelle auf 10.000 Zeilen begrenzt. Das Original kann nur über die Datasources-Administrations-API mit uneingeschränkter aktueller und historischer Sichtbarkeit abgerufen werden, nicht über Katalogmetadaten. "
                "Fehlende ältere Originale lassen sich nicht rekonstruieren. Der Textmodus kann Spalten zu Zeichenketten machen; prüfen Sie deshalb nachgelagerte Zahlenvergleiche und Umwandlungen vor der Übernahme."
            ),
        }},
    ),
    DocumentationTopic(
        id="dataflow.save-completion",
        title="Editing while a pipeline save completes",
        summary="Keep newer local edits separate from the immutable revision accepted by the server.",
        body=(
            "You may continue editing a pipeline while Save is pending. The accepted server revision becomes the saved baseline; "
            "fields changed since submission remain in the local unsaved draft and require another explicit save. Graphs are kept as whole values, "
            "not merged or reordered node by node. Save-and-leave does not navigate while newer edits remain unsaved. "
            "A second save uses the revision actually accepted by the first request; duplicate concurrent save submissions are blocked. "
            "Selecting, replacing or discarding a draft, leaving the page, or changing authentication context prevents an old completion from replacing the current editor. "
            "Such a request may already have succeeded on the server: reload and review before retrying if the context changed. "
            "Ordinary session refreshes with the same identity, credentials and permissions keep accepted IDs and revisions; cosmetic profile changes do not interrupt saving. "
            "If a save was accepted across a real authorization change, further saves in that edit session are blocked until the draft is replaced after review, so a new pipeline is not created twice. "
            "Revision conflicts retain the local draft and require review; neither the UI nor administrators automatically overwrite a conflicting server revision. "
            "Current permissions and governance remain server-enforced. No preview rows are persisted by this editor behavior."
        ),
        layer="always", documentation_types=("user", "admin"), audience=("user", "module_admin", "operator"), order=7,
        translations={"de": {
            "title": "Während des Speicherns einer Pipeline weiterarbeiten",
            "summary": "Neuere lokale Änderungen von der unveränderlichen, serverseitig angenommenen Revision trennen.",
            "body": (
                "Während Speichern läuft, können Sie die Pipeline weiter bearbeiten. Die angenommene Serverrevision wird zum gespeicherten Vergleichsstand; "
                "seit dem Absenden geänderte Felder bleiben im lokalen, ungespeicherten Entwurf und benötigen einen weiteren ausdrücklichen Speichervorgang. "
                "Graphen bleiben vollständige Werte und werden nicht knotenweise zusammengeführt oder umsortiert. Speichern und Verlassen navigiert nicht, solange neuere Änderungen ungespeichert sind. "
                "Ein zweiter Speichervorgang verwendet die tatsächlich angenommene Revision des ersten; doppelte gleichzeitige Speicheranfragen werden blockiert. "
                "Auswahl, Ersetzen oder Verwerfen eines Entwurfs, Verlassen der Seite oder ein geänderter Authentifizierungskontext verhindern, dass ein altes Ergebnis den aktuellen Editor ersetzt. "
                "Die Anfrage kann auf dem Server bereits erfolgreich gewesen sein: Nach einem Kontextwechsel vor einem erneuten Versuch neu laden und prüfen. "
                "Gewöhnliche Sitzungsaktualisierungen mit gleicher Identität, gleichen Zugangsdaten und Rechten behalten angenommene IDs und Revisionen; rein optische Profiländerungen unterbrechen das Speichern nicht. "
                "Wurde ein Speichervorgang während einer tatsächlichen Berechtigungsänderung angenommen, bleiben weitere Speicheranfragen dieser Bearbeitungssitzung bis zum geprüften Ersetzen des Entwurfs gesperrt, damit keine Pipeline doppelt entsteht. "
                "Bei Revisionskonflikten bleibt der lokale Entwurf erhalten und muss geprüft werden; weder Oberfläche noch Administratoren überschreiben automatisch eine widersprechende Serverrevision. "
                "Aktuelle Rechte und Governance werden weiterhin serverseitig geprüft. Dieses Editorverhalten speichert keine Vorschauzeilen dauerhaft."
            ),
        }},
    ),
    DocumentationTopic(
        id="dataflow.reference-worker-limits",
        title="Reference execution process limits",
        summary="Contain expensive expressions and intermediate allocations without changing datasource authority.",
        body=(
            "Reference previews and reference development runs evaluate in a fresh disposable process, including regex and aggregate intermediate allocations. "
            "The existing row, node and per-result byte checks remain. The process additionally enforces the request's wall-clock and memory budgets "
            "(default preview: 2 seconds and 256 MiB virtual address space), rounded-up CPU seconds, no regular-file output, and 32 MiB per data-only input/result transport. "
            "Supported process budgets are at most 600 seconds and 8 GiB; unsupported controls or exceeded limits fail with structured backend.process diagnostics, never inline fallback. "
            "Datasource authorization and bounded source reads remain in the parent, including nested subflow sources; sessions and credentials are not passed to the child. "
            "Reference source collection also checks a cumulative 32 MiB typed-data budget before constructing another columnar copy; providers retain their separate per-read limits. "
            "Completed node diagnostics survive ordinary evaluation errors; a killed worker returns no partial rows or invented node progress. "
            "GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY limits shared isolated-work admission per API/worker process, default 1; busy capacity is retryable. "
            "It is not a fleet-wide quota or arbitrary-code sandbox. Cancellation checks before/after reference runs remain; hard wall limits stop an unresponsive expression. "
            "Staging/production still require the separate DuckDB backend and are not converted to reference execution."
        ),
        layer="static", documentation_types=("user", "admin"), audience=("user", "module_admin", "operator"), order=6,
        translations={"de": {
            "title": "Prozessgrenzen der Referenzausführung",
            "summary": "Aufwendige Ausdrücke und Zwischenspeicher begrenzen, ohne Datenquellenrechte zu verändern.",
            "body": (
                "Referenz-Vorschauen und Referenz-Entwicklungsläufe werten Ausdrücke in einem frischen, kurzlebigen Prozess aus, einschließlich regulärer Ausdrücke und großer Zwischenergebnisse. "
                "Bestehende Zeilen-, Knoten- und Ergebnis-Bytegrenzen bleiben bestehen. Zusätzlich gelten das Laufzeit- und Speicherbudget der Anfrage "
                "(Vorschau standardmäßig 2 Sekunden und 256 MiB virtueller Adressraum), aufgerundete CPU-Sekunden, keine regulären Ausgabedateien und je 32 MiB für den reinen Datentransport. "
                "Prozessbudgets unterstützen höchstens 600 Sekunden und 8 GiB. Fehlende Betriebssystemkontrollen oder überschrittene Grenzen erzeugen strukturierte backend.process-Diagnosen, ohne Ausweichbetrieb im Hauptprozess. "
                "Datenquellenrechte und begrenzte Quellabrufe werden im Hauptprozess geprüft, auch für verschachtelte Teilflüsse; Sitzungen und Zugangsdaten gelangen nicht in den Kindprozess. "
                "Referenz-Quellabrufe prüfen außerdem zusammen höchstens 32 MiB typisierte Daten, bevor eine weitere spaltenweise Kopie entsteht; getrennte Abrufgrenzen der Anbieter bleiben bestehen. "
                "Gewöhnliche Auswertungsfehler behalten bereits abgeschlossene Knotendiagnosen. Ein gestoppter Prozess liefert keine Teilzeilen und keinen erfundenen Knotenfortschritt. "
                "GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY begrenzt gemeinsam genutzte isolierte Arbeit je API-/Worker-Prozess, standardmäßig 1; bei Auslastung ist ein erneuter Versuch möglich. "
                "Dies ist weder eine systemweite Quote noch eine Sandbox für beliebigen Code. Abbruchprüfungen vor und nach Referenzläufen bleiben erhalten; harte Laufzeitgrenzen stoppen hängende Ausdrücke. "
                "Staging und Produktion benötigen weiterhin das gesonderte DuckDB-Backend und wechseln nicht zur Referenzausführung."
            ),
        }},
    ),
    DocumentationTopic(
        id="dataflow.workspace-layout",
        title="Dataflow workspace actions",
        summary="Find collection-wide commands in their consistent workspace position.",
        body="The workspace documentation book sits beside Pipelines; automation and run help sits beside "
        "the corresponding dialog title, and field help stays with its label. "
        "Reload and New pipeline use the persistent full-width workspace header at the upper right; Reload sits immediately before creation. Selecting a record, changing filters, or opening an editor does not move these collection-wide commands into the left pane. SQL editing, validation, previews, saving, and execution keep their existing editor scope and bounded safety rules. Existing permissions, disabled-state rules, and unsaved-change guards still apply. Administrators configure authority through the existing permission system; no new permission or automatic operation is introduced.",
        layer="static",
        documentation_types=("user", "admin"),
        audience=("user", "module_admin", "operator"),
        order=5,
        translations={"de": {
            "title": "Datenflüsse: Aktionen im Arbeitsbereich",
            "summary": "Sammlungsweite Aktionen an ihrer einheitlichen Position im Arbeitsbereich finden.",
            "body": "Das Dokumentationsbuch des Arbeitsbereichs steht neben Pipelines; Hilfe zu Automatisierung "
            "und Ausführung steht neben dem jeweiligen Dialogtitel, und Feldhilfe bleibt bei der "
            "Feldbezeichnung. "
            "Neu laden und Neue Pipeline stehen oben rechts in der dauerhaft sichtbaren, arbeitsbereichsweiten Leiste; Neu laden steht unmittelbar vor dem Anlegen. Auswahl, Filterwechsel und Bearbeitung verschieben diese sammlungsweiten Aktionen nicht in den linken Bereich. SQL-Bearbeitung, Validierung, Vorschau, Speichern und Ausführung behalten ihren bisherigen Editorbereich und ihre begrenzenden Sicherheitsregeln. Bestehende Berechtigungen, Deaktivierungsregeln und der Schutz ungespeicherter Änderungen gelten weiterhin. Administratoren konfigurieren Rechte im bestehenden Berechtigungssystem; es entstehen weder neue Rechte noch automatische Vorgänge.",
        }},
    ),
    DocumentationTopic(
        id="dataflow.data-subject-requests",
        title="Dataflow data-subject requests",
        summary="Minimize retained transformation detail without treating derived flows as authoritative subject records.",
        body=(
            "Dataflow matches exact tenant-scoped pipeline, revision, reconciliation, run, deployment, trigger, and delivery identifiers plus minimized account, identity, and membership attribution. Results never copy graphs, SQL, request or event payloads, reconciliation corrections, authorization snapshots, provenance bodies, errors, source details, hashes, credentials, or output rows. Dataflow does not scan arbitrary transformation content for a person; the authoritative input module must locate and correct subject facts. "
            "Explicitly identified terminal run and delivery detail can be minimized idempotently, and automation authority linked to the subject can be disabled and revoked. Definitions, reconciliation evidence, active work, deployments, broad pipeline packages, published Datasource outputs, and institutional attribution require authorized review or retention. Correct sources and refresh Datasource, Search, and Reporting derivatives after review."
        ),
        layer="configured",
        documentation_types=("admin", "user"),
        audience=("user", "operator", "module_admin", "data_steward", "auditor"),
        order=74,
        related_modules=("core", "datasources", "reporting", "workflow_engine"),
        metadata={
            "help_contexts": [
                "dataflow.data-subject-requests",
                "dataflow.runs",
                "dataflow.triggers",
            ],
        },
    ),
    DocumentationTopic(
        id="dataflow.module-boundary",
        title="Dataflow module boundary",
        summary="Versioned tabular transformations with graphical and constrained SQL editing.",
        body=(
            "Dataflow owns canonical pipeline graphs, immutable revisions, validation, constrained "
            "SQL compilation, preview and run diagnostics, and lineage references. Datasources owns "
            "the governed catalogue and materializations, while Connectors owns external acquisition "
            "and credentials; Reporting owns analytical presentation and exports; "
            "Workflow owns orchestration and human handoffs; Risk Compliance owns sanctions review "
            "semantics and policy gates. User SQL is compiled into approved transforms and is never "
            "passed unchecked to a backing database. Opening or reloading Dataflow does not "
            "start a pipeline. If the editor cannot be loaded after a development update, "
            "preserve unsaved work before reloading the browser. Administrators should "
            "distinguish frontend asset failures from pipeline API or access errors. "
            "Development and browser-conformance servers use separate dependency caches; "
            "an older server may need restarting after this configuration update."
        ),
        layer="available",
        documentation_types=("admin", "user"),
        audience=("operator", "module_admin", "power_user", "product_owner"),
        order=75,
        related_modules=(
            "datasources",
            "connectors",
            "files",
            "reporting",
            "workflow_engine",
            "risk_compliance",
            "notifications",
            "policy",
            "audit",
        ),
        metadata={
            "kind": "workflow",
            "route": "/dataflow",
            "screen": "Dataflow",
            "prerequisites": [
                "You may read Dataflow pipeline definitions in the active tenant.",
            ],
            "steps": [
                "Open Dataflow and select a pipeline or create an authorized draft.",
                "Inspect the graph, immutable revision, diagnostics, and declared source references.",
                "Use preview or run actions only when the effective permissions and Policy allow them.",
            ],
            "outcome": "The pipeline remains a governed transformation definition with explicit module boundaries and source authority.",
            "verification": "Confirm the active revision, validation diagnostics, source references, and permitted actions in the Dataflow workspace.",
            "first_slice": (
                "Inline and governed datasources, union, join, filter, deduplication, select, "
                "typed expressions, conversion, quality and reconciliation, reusable subflows, "
                "aggregate, sort, limit, output, revisioning, and bounded preview."
            ),
            "sql_safety": "Constrained AST compilation only; no pass-through execution.",
            "run_recovery": (
                "Database-only runs commit atomically with Core recovery evidence. "
                "Output publication uses forward recovery and blocks blind retry "
                "when provider acknowledgement is uncertain."
            ),
            "help_contexts": [
                "dataflow.page",
                "dataflow.library",
                "dataflow.graph",
                "dataflow.sql",
                "dataflow.inspector",
                "dataflow.results",
                "dataflow.state.read-only",
            ],
        },
        conditions=(
            DocumentationCondition(
                required_modules=("dataflow",),
                required_scopes=(READ_SCOPE,),
            ),
        ),
        structured_translation_version="1",
        structured_translations={
            "de": {
                "prerequisites": [
                    "Sie dürfen Dataflow-Pipeline-Definitionen im aktiven Mandanten lesen.",
                ],
                "steps": [
                    "Öffnen Sie Dataflow und wählen Sie eine Pipeline oder legen Sie einen autorisierten Entwurf an.",
                    "Prüfen Sie Graph, unveränderliche Revision, Diagnosen und ausgewiesene Quellreferenzen.",
                    "Verwenden Sie Vorschau- oder Laufaktionen nur, wenn wirksame Berechtigungen und Policy sie erlauben.",
                ],
                "outcome": "Die Pipeline bleibt eine gesteuerte Transformationsdefinition mit ausdrücklichen Modulgrenzen und Quellenautorität.",
                "verification": "Prüfen Sie aktive Revision, Validierungsdiagnosen, Quellreferenzen und erlaubte Aktionen im Dataflow-Arbeitsbereich.",
            }
        },
    ),
    DocumentationTopic(
        id="dataflow.reference.nodes-and-expressions",
        title="Dataflow nodes and expressions",
        summary="Typed node inputs, expressions, schema propagation, and bounded intermediate previews.",
        body=(
            "Every graph node declares typed inputs, configuration, output schema, and validation rules. "
            "Source nodes pin inline content or governed Datasource references; combine, filter, transform, "
            "quality, reconciliation, reusable-subflow, and output nodes remain explicit in the canonical graph. "
            "Reusable subflows select a Policy-authorized immutable flow or template revision. The server resolves "
            "and pins its graph, source hash, Policy decision, and closed typed input/output contracts; caller-supplied "
            "graph snapshots are ignored, incompatible inputs fail validation, and nested reference cycles are rejected. "
            "Reconciliation rows expose stable key hashes, explicit before/after values, and input hashes. The "
            "review dialog records accept, reject, correct, or defer decisions in tenant-owned immutable decision "
            "sets. Their current projection is a fingerprinted Dataflow source; every superseded revision retains "
            "the actor, reason, exact input hash, time, and optional correction. A reconcile.decisions node can "
            "annotate exact matches, invalidate changed inputs, and report orphaned decisions without silently "
            "rewriting business data. "
            "Expressions use the typed Dataflow expression language and never execute arbitrary host or database "
            "code. Selecting a node may request a bounded intermediate preview; preview rows are transient, "
            "privacy-filtered for the actor, and are not retained as run output. SQL editing compiles into the same "
            "canonical graph, so unsupported statements are diagnostics rather than pass-through SQL. "
            "Reference previews retain the existing 1,000,000-byte serialized result limit per node. "
            "LPAD and RPAD reject target lengths above 1,000,000 characters before allocating padding, "
            "including oversized intermediate values inside LENGTH or SUBSTRING even if the final scalar would be small. "
            "Reduce the requested padding length; this fails the expression at its node instead of truncating data. "
            "Ordinary Unicode padding, null inputs, and in-budget truncation keep their existing behavior; "
            "the final byte check still accounts for JSON and multibyte character overhead. "
            "This allocation guard does not replace runtime or byte limits for other expression operations."
        ),
        layer="available",
        documentation_types=("admin", "user"),
        audience=("operator", "module_admin", "power_user", "data_steward"),
        order=76,
        related_modules=("datasources", "connectors", "policy", "audit"),
        metadata={
            "kind": "reference",
            "help_contexts": [
                "dataflow.field.node-name",
                "dataflow.field.source",
                "dataflow.field.expression",
                "dataflow.field.schema",
                "dataflow.field.reusable-input-binding",
                "dataflow.field.subflow-reference",
                "dataflow.field.subflow-revision",
                "dataflow.action.preview-node",
                "dataflow.action.review-decisions",
            ],
        },
    ),
    DocumentationTopic(
        id="dataflow.reference.fields-and-consequences",
        title="Dataflow fields and lifecycle consequences",
        summary="Definition scope, revision, reuse, automation, execution, publication, promotion, and deletion semantics.",
        body=(
            "Scope determines ownership and Policy inheritance. Templates can be derived but not run; complete "
            "flows may be previewed, revisioned, automated, and executed when effective Policy allows it. Saving "
            "appends an immutable revision. A scoped copy pins its source revision and content hash. A newer "
            "source revision is reported without changing the copy. Adopting it requires an exact reviewed source "
            "hash and a reason, appends an immutable copy revision, records the Policy decision and reviewer, and "
            "returns the copy to draft so runs and automation cannot use the changed graph before activation. Triggers pin "
            "the revision and authorization grant, then re-evaluate authority for every delivery. Allow runs is "
            "the definition-level admission boundary and does not grant a caller permission. A one-time run uses "
            "the configured tenant-local date and time. The missed-run policy either coalesces elapsed interval "
            "occurrences into one latest delivery or skips them; it never silently replays every missed occurrence. "
            "The concurrency limit bounds active deliveries for that trigger and does not increase tenant worker "
            "capacity. Runs create "
            "durable command and recovery evidence. Publishing creates a governed Datasource materialization, and "
            "environment promotion changes which immutable revision is eligible for staging or production runs. "
            "Recording a reconciliation decision uses optimistic concurrency and appends an immutable revision; "
            "it never mutates the reviewed business row. "
            "Deletion prevents future use while retained run, deployment, lineage, audit, and recovery evidence "
            "continues under its retention policy."
        ),
        layer="available",
        documentation_types=("admin", "user"),
        audience=("operator", "module_admin", "power_user", "product_owner"),
        order=77,
        related_modules=(
            "datasources",
            "workflow_engine",
            "notifications",
            "policy",
            "audit",
        ),
        metadata={
            "kind": "reference",
            "help_contexts": [
                "dataflow.field.scope",
                "dataflow.field.definition-kind",
                "dataflow.field.allow-runs",
                "dataflow.field.trigger-run-at",
                "dataflow.field.trigger-missed-runs",
                "dataflow.field.trigger-concurrency",
                "dataflow.field.rebase-reason",
                "dataflow.action.save",
                "dataflow.action.derive",
                "dataflow.action.rebase",
                "dataflow.action.trigger",
                "dataflow.action.record-decision",
                "dataflow.action.delete",
            ],
            "consequence_classes": {
                "save_revision": "Appends an immutable pipeline definition revision.",
                "derive_copy": "Creates a separately governed copy pinned to the source revision and hash.",
                "rebase_copy": "Appends the exact reviewed source revision to a scoped copy, records reviewer provenance, and returns it to draft.",
                "configure_trigger": "Creates or changes an automation command with revision and authorization evidence.",
                "record_decision": "Appends an actor-attributed immutable decision revision against an exact input hash.",
                "delete_pipeline": "Prevents future use while retained evidence remains governed.",
            },
        },
    ),
    DocumentationTopic(
        id="dataflow.execution-and-recovery",
        title="Dataflow execution, publication, and recovery",
        summary="Pinned runs, environment promotion, output publication, cancellation, reconciliation, and retained evidence.",
        body=(
            "Every run is pinned to an immutable revision and idempotency key. The queue records actor, authority, "
            "environment, progress, cancellation, output, and recovery state. Database-only runs commit atomically. "
            "Publication to a governed Datasource uses forward recovery: an unknown provider outcome is reconciled "
            "before retry so output is not duplicated. Staging and production promotion is explicit and does not "
            "rewrite a revision. Freezing a published state assigns a durable label to the exact immutable output; "
            "it does not copy or detach the data from Datasources retention, hold, and access rules. Artifact-backed "
            "outputs return the same stable publication, datasource, and materialization references as inline "
            "outputs. Datasource warnings and review-required states remain visible to Workflow instead of being "
            "collapsed into success. Cancellation is best effort once external work has started; the final evidence "
            "states whether work stopped, completed, failed, or requires operator reconciliation. Scheduled, event, "
            "and queued execution is partitioned by tenant module entitlement before a run is claimed. Disabling "
            "Dataflow stops new admission and leaves accepted runs available for an explicit operator decision."
            " Reporting may pin a successful published run; Dataflow then rechecks run authority and Datasource access and reads only the exact recorded materialization without reparameterizing or re-executing it."
        ),
        layer="available",
        documentation_types=("admin", "user"),
        audience=("operator", "module_admin", "power_user", "security_admin"),
        order=78,
        related_modules=("datasources", "notifications", "policy", "audit", "ops"),
        metadata={
            "help_contexts": [
                "dataflow.runs",
                "dataflow.action.queue-run",
                "dataflow.action.publish",
                "dataflow.field.freeze-publication",
                "dataflow.action.promote-staging",
                "dataflow.action.promote-production",
                "dataflow.state.recovery-attention",
            ],
            "consequence_classes": {
                "queue_run": "Creates a durable asynchronous command and authorization evidence.",
                "publish_output": "Creates or updates a governed Datasource and appends a materialization.",
                "promote_revision": "Makes an immutable revision eligible in a higher execution environment.",
                "cancel_run": "Requests cancellation; already acknowledged external effects may remain.",
            },
        },
    ),
))


def _dataflow_router(context: ModuleContext):
    from govoplan_dataflow.backend.runtime import configure_runtime

    configure_runtime(registry=context.registry, settings=context.settings)
    from govoplan_dataflow.backend.router import router

    return router


def _run_provider(context: ModuleContext):
    from govoplan_dataflow.backend.service import SqlDataflowRunLifecycleProvider

    return SqlDataflowRunLifecycleProvider(registry=context.registry)


def _dsar_provider(context: ModuleContext) -> DataflowDsarProvider:
    del context
    return DataflowDsarProvider()


def _run_worker(context: ModuleContext):
    from govoplan_dataflow.backend.run_worker import SqlDataflowRunWorker

    return SqlDataflowRunWorker(registry=context.registry)


def _trigger_provider(context: ModuleContext):
    from govoplan_dataflow.backend.triggers import (
        SqlDataflowTriggerDispatcher,
    )

    return SqlDataflowTriggerDispatcher(registry=context.registry)


def _search_source_provider(context: ModuleContext):
    from govoplan_dataflow.backend.search_source import (
        create_dataflow_search_source,
    )

    return create_dataflow_search_source(context)


def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
    return {
        "dataflow_pipelines": (
            session.query(dataflow_models.DataflowPipeline)
            .filter(
                dataflow_models.DataflowPipeline.tenant_id == tenant_id,
                dataflow_models.DataflowPipeline.deleted_at.is_(None),
            )
            .count()
        ),
        "dataflow_runs": (
            session.query(dataflow_models.DataflowRun)
            .filter(dataflow_models.DataflowRun.tenant_id == tenant_id)
            .count()
        ),
        "dataflow_triggers": (
            session.query(dataflow_models.DataflowTrigger)
            .filter(dataflow_models.DataflowTrigger.tenant_id == tenant_id)
            .count()
        ),
        "dataflow_trigger_deliveries": (
            session.query(dataflow_models.DataflowTriggerDelivery)
            .filter(dataflow_models.DataflowTriggerDelivery.tenant_id == tenant_id)
            .count()
        ),
        "dataflow_reconciliation_decision_sets": (
            session.query(dataflow_models.DataflowReconciliationDecisionSet)
            .filter(
                dataflow_models.DataflowReconciliationDecisionSet.tenant_id == tenant_id
            )
            .count()
        ),
        "dataflow_reconciliation_decisions": (
            session.query(dataflow_models.DataflowReconciliationDecision)
            .filter(
                dataflow_models.DataflowReconciliationDecision.tenant_id == tenant_id
            )
            .count()
        ),
    }


manifest = ModuleManifest(
    id=MODULE_ID,
    name=MODULE_NAME,
    version=MODULE_VERSION,
    dependencies=(),
    optional_dependencies=(
        "access",
        "audit",
        "datasources",
        "files",
        "notifications",
        "policy",
        "reporting",
        "risk_compliance",
        "search",
        "workflow_engine",
    ),
    optional_capabilities=(
        CAPABILITY_ACCESS_DIRECTORY,
        CAPABILITY_ACCESS_REFERENCE_OPTIONS,
        CAPABILITY_DATASOURCE_CATALOGUE,
        CAPABILITY_DATASOURCE_LIFECYCLE,
        CAPABILITY_DATASOURCE_PUBLICATION,
        CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
        CAPABILITY_NOTIFICATIONS_DISPATCH,
        CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
    ),
    provides_interfaces=(
        ModuleInterfaceProvider(
            name="dataflow.pipeline_catalog", version=MODULE_VERSION
        ),
        ModuleInterfaceProvider(
            name="dataflow.pipeline_preview", version=MODULE_VERSION
        ),
        ModuleInterfaceProvider(name="dataflow.run_lifecycle", version=MODULE_VERSION),
        ModuleInterfaceProvider(name="dataflow.run_worker", version=MODULE_VERSION),
        ModuleInterfaceProvider(
            name=CAPABILITY_DATAFLOW_DATASET_OUTPUT, version=MODULE_VERSION
        ),
        ModuleInterfaceProvider(
            name="dataflow.trigger_dispatcher",
            version=MODULE_VERSION,
        ),
        ModuleInterfaceProvider(name=DATAFLOW_DSAR_CAPABILITY, version="0.1.0"),
    ),
    requires_interfaces=(
        ModuleInterfaceRequirement(
            name=CAPABILITY_ACCESS_REFERENCE_OPTIONS,
            version_min="0.1.0",
            version_max_exclusive="0.2.0",
            optional=True,
        ),
        ModuleInterfaceRequirement(
            name="datasources.catalogue",
            version_min="0.1.0",
            version_max_exclusive="1.0.0",
            optional=True,
        ),
        ModuleInterfaceRequirement(
            name="policy.definition_governance",
            version_min="0.1.0",
            version_max_exclusive="1.0.0",
            optional=True,
        ),
        ModuleInterfaceRequirement(
            name="auth.automation_principal",
            version_min="0.1.0",
            version_max_exclusive="1.0.0",
            optional=True,
        ),
        ModuleInterfaceRequirement(
            name="datasources.lifecycle",
            version_min="0.1.0",
            version_max_exclusive="1.0.0",
            optional=True,
        ),
        ModuleInterfaceRequirement(
            name="datasources.publication",
            version_min="0.1.0",
            version_max_exclusive="1.0.0",
            optional=True,
        ),
        ModuleInterfaceRequirement(
            name=CAPABILITY_NOTIFICATIONS_DISPATCH,
            version_min="0.1.0",
            version_max_exclusive="1.0.0",
            optional=True,
        ),
        ModuleInterfaceRequirement(
            name="search.source",
            version_min="1.0.0",
            version_max_exclusive="2.0.0",
            optional=True,
        ),
    ),
    permissions=PERMISSIONS,
    role_templates=ROLE_TEMPLATES,
    nav_items=(
        NavItem(
            path="/dataflow",
            label="Dataflow",
            icon="waypoints",
            required_any=(READ_SCOPE, ADMIN_SCOPE),
            order=72,
        ),
    ),
    frontend=FrontendModule(
        module_id=MODULE_ID,
        package_name="@govoplan/dataflow-webui",
        routes=(
            FrontendRoute(
                path="/dataflow",
                component="DataflowPage",
                required_any=(READ_SCOPE, ADMIN_SCOPE),
                order=72,
            ),
        ),
        nav_items=(
            NavItem(
                path="/dataflow",
                label="Dataflow",
                icon="waypoints",
                required_any=(READ_SCOPE, ADMIN_SCOPE),
                order=72,
            ),
        ),
        product_areas=(
            ProductAreaContribution(
                id="data-assurance",
                module_id=MODULE_ID,
                label="i18n:govoplan-core.product_area.data_assurance",
                icon="database-zap",
                description="i18n:govoplan-core.product_area.data_assurance_description",
                surface_ids=("dataflow.nav.dataflow", "dataflow.route.dataflow"),
                order=60,
            ),
        ),
        view_surfaces=(
            ViewSurface(
                id="dataflow.page",
                module_id=MODULE_ID,
                kind="route",
                label="Dataflow",
                order=72,
            ),
            ViewSurface(
                id="dataflow.library",
                module_id=MODULE_ID,
                kind="section",
                label="Pipeline library",
                parent_id="dataflow.page",
                order=10,
            ),
            ViewSurface(
                id="dataflow.graph",
                module_id=MODULE_ID,
                kind="section",
                label="Graph editor",
                parent_id="dataflow.page",
                order=20,
            ),
            ViewSurface(
                id="dataflow.sql",
                module_id=MODULE_ID,
                kind="section",
                label="Constrained SQL editor",
                parent_id="dataflow.page",
                order=30,
            ),
            ViewSurface(
                id="dataflow.inspector",
                module_id=MODULE_ID,
                kind="section",
                label="Node inspector",
                parent_id="dataflow.page",
                order=40,
            ),
            ViewSurface(
                id="dataflow.results",
                module_id=MODULE_ID,
                kind="section",
                label="Preview and diagnostics",
                parent_id="dataflow.page",
                order=50,
            ),
            ViewSurface(
                id="dataflow.decisions",
                module_id=MODULE_ID,
                kind="action",
                label="Reconciliation decisions",
                parent_id="dataflow.results",
                order=55,
            ),
            ViewSurface(
                id="dataflow.triggers",
                module_id=MODULE_ID,
                kind="action",
                label="Automation triggers",
                parent_id="dataflow.page",
                order=60,
            ),
            ViewSurface(
                id="dataflow.runs",
                module_id=MODULE_ID,
                kind="action",
                label="Runs and deployments",
                parent_id="dataflow.page",
                order=70,
            ),
            ViewSurface(
                id="dataflow.widget.pipelines",
                module_id=MODULE_ID,
                kind="section",
                label="Dataflows widget",
                order=70,
            ),
        ),
    ),
    route_factory=_dataflow_router,
    capability_factories={
        CAPABILITY_DATAFLOW_DATASET_OUTPUT: lambda context: __import__(
            "govoplan_dataflow.backend.dataset_output",
            fromlist=["dataset_output_provider"],
        ).dataset_output_provider(context),
        CAPABILITY_DATAFLOW_RUN_LIFECYCLE: _run_provider,
        CAPABILITY_DATAFLOW_RUN_WORKER: _run_worker,
        CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER: _trigger_provider,
        DATAFLOW_DSAR_CAPABILITY: _dsar_provider,
    },
    capability_documentation={
        DATAFLOW_DSAR_CAPABILITY: CapabilityDocumentation(
            label="Dataflow data-subject request provider",
            summary="Finds and minimizes subject-linked transformation state and automation authority.",
            contract_version="0.1.0",
            documentation_types=("admin", "user"),
            audience=("privacy_officer", "data_steward", "user"),
        ),
    },
    search_sources=(
        SearchSourceProviderRegistration(
            id="dataflow.pipelines",
            factory=_search_source_provider,
        ),
    ),
    tenant_summary_providers=(_tenant_summary,),
    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(
            dataflow_models.DataflowReconciliationDecision,
            dataflow_models.DataflowReconciliationDecisionSet,
            dataflow_models.DataflowTriggerDelivery,
            dataflow_models.DataflowTrigger,
            dataflow_models.DataflowRun,
            dataflow_models.DataflowPipelineDeployment,
            dataflow_models.DataflowPipelineRevision,
            dataflow_models.DataflowPipeline,
            label="Dataflow",
        ),
        retirement_notes=(
            "Destructive retirement drops Dataflow definitions, revisions, and run evidence "
            "after the installer captures a database snapshot."
        ),
    ),
    uninstall_guard_providers=(
        persistent_table_uninstall_guard(
            dataflow_models.DataflowPipeline,
            dataflow_models.DataflowPipelineRevision,
            dataflow_models.DataflowReconciliationDecisionSet,
            dataflow_models.DataflowReconciliationDecision,
            dataflow_models.DataflowPipelineDeployment,
            dataflow_models.DataflowRun,
            dataflow_models.DataflowTrigger,
            dataflow_models.DataflowTriggerDelivery,
            label="Dataflow",
        ),
    ),
    documentation=DOCUMENTATION,
    architecture=declared_module_architecture(
        layer="data_reporting_integration",
        kind="runtime",
        maturity="vertical_slice",
        documentation_ref="README.md",
        test_ref="tests/test_golden_flows.py",
        known_limits=(
            "Execution adapters do not yet cover every declared node family.",
        ),
        owned_concepts=(
            "dataflow definition",
            "dataflow revision",
            "dataflow run",
            "reconciliation decision set",
            "transformation graph",
        ),
        non_owned_concepts=(
            "datasource binding",
            "connector transport",
            "report presentation",
            "workflow task",
        ),
        recovery_docs=("README.md", "docs/DURABLE_RUN_RECOVERY.md"),
        security_docs=("README.md",),
        operations_docs=("README.md",),
    ),
)


manifest = with_documentation_structured_translations(
    manifest, locale="de", translations=GERMAN_STRUCTURED_TRANSLATIONS
)


def get_manifest() -> ModuleManifest:
    return manifest


__all__ = [
    "ADMIN_SCOPE",
    "MODULE_ID",
    "MODULE_NAME",
    "MODULE_VERSION",
    "READ_SCOPE",
    "RUN_SCOPE",
    "TRIGGER_DISPATCH_SCOPE",
    "TRIGGER_READ_SCOPE",
    "TRIGGER_WRITE_SCOPE",
    "WRITE_SCOPE",
    "get_manifest",
    "manifest",
]