Neo4j Constraint Compliance Gate

id: neo4j-constraint-compliance-gate
namespace: company.team
description: |
  Check whether a target node label's governance-critical property is backed
  by a Neo4j uniqueness constraint using SHOW CONSTRAINTS, alert Slack on a
  gap (AT_RISK) or an unpopulated label (UNREACHABLE), and optionally create
  the missing constraint with CREATE CONSTRAINT ... IF NOT EXISTS.
  auto_remediate and dry_run are two independent gates: existing node data is
  never read, modified, or deleted - only schema metadata is touched.

triggers:
  - id: periodic_constraint_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check every 6 hours - constraint schema drifts slowly compared to
      data freshness, so this does not need a tighter cadence. Shipped disabled;
      enable it once neo4j_url/target_label point at a real instance and a
      manual run has been verified. Always fires in safe mode (auto_remediate
      false, dry_run true) regardless of this flow's own defaults.
    cron: "0 */6 * * *"
    disabled: true
    inputs:
      auto_remediate: "false"
      dry_run: "true"

inputs:
  - id: neo4j_url
    type: STRING
    displayName: "Neo4j Bolt URL"
    defaults: "bolt://localhost:7687"
    description: Bolt endpoint passed to io.kestra.plugin.neo4j.Query's url property.

  - id: target_label
    type: SELECT
    displayName: "Target Node Label"
    defaults: "Account"
    values:
      - "Account"
      - "Device"
      - "Merchant"
    description: Node label audited for constraint coverage. Bounded to a fixed set
      so the label is never spliced into Cypher from unbounded input.

  - id: required_unique_property
    type: STRING
    displayName: "Required Unique Property"
    defaults: "account_id"
    description: Property expected to be covered by a node uniqueness constraint on
      target_label. Compared as a string value inside a Cypher WHERE clause,
      never used to build the query's structure.

  - id: auto_remediate
    type: SELECT
    displayName: "Auto Remediate"
    defaults: "false"
    values:
      - "false"
      - "true"
    description: If "true" and target_label is populated but the constraint is
      missing, create it with CREATE CONSTRAINT ... IF NOT EXISTS (still gated
      by dry_run below). If "false", the gap is only reported.

  - id: dry_run
    type: SELECT
    displayName: "Dry Run"
    defaults: "true"
    values:
      - "true"
      - "false"
    description: If "true", remediation is only logged, CREATE CONSTRAINT never
      runs. Must be explicitly "false", together with auto_remediate "true", for
      the constraint to actually be created.

tasks:
  - id: check_label_population
    type: io.kestra.plugin.neo4j.Query
    description: Counts nodes carrying target_label. A zero count means the label is
      not populated in this graph - there is nothing to govern, and a
      constraint-coverage result would be meaningless, so this is checked before
      the constraint query.
    url: "{{ inputs.neo4j_url }}"
    username: "{{ secret('NEO4J_USERNAME') }}"
    password: "{{ secret('NEO4J_PASSWORD') }}"
    query: "MATCH (n:{{ inputs.target_label }}) RETURN count(n) AS node_count"
    storeType: FETCHONE

  - id: check_constraint_coverage
    type: io.kestra.plugin.neo4j.Query
    description: SHOW CONSTRAINTS is Neo4j's own schema metadata command, YIELD-ed
      and filtered to node-type constraints on target_label that cover
      required_unique_property, then counted server-side.
    url: "{{ inputs.neo4j_url }}"
    username: "{{ secret('NEO4J_USERNAME') }}"
    password: "{{ secret('NEO4J_PASSWORD') }}"
    query: "SHOW CONSTRAINTS YIELD name, type, entityType, labelsOrTypes, properties
      WHERE entityType = 'NODE' AND '{{ inputs.target_label }}' IN labelsOrTypes
      AND '{{ inputs.required_unique_property }}' IN properties RETURN count(*)
      AS matching_constraints"
    storeType: FETCHONE

  - id: evaluate_population
    type: io.kestra.plugin.core.flow.If
    description: UNREACHABLE when target_label has zero nodes - distinct from
      AT_RISK, where the label exists but lacks the required constraint.
    condition: "{{ outputs.check_label_population.row.node_count > 0 }}"
    then:
      - id: evaluate_constraint_presence
        type: io.kestra.plugin.core.flow.If
        description: AT_RISK when the label is populated but no matching constraint
          covers required_unique_property.
        condition: "{{ outputs.check_constraint_coverage.row.matching_constraints > 0 }}"
        then:
          - id: log_constraint_healthy
            type: io.kestra.plugin.core.log.Log
            message: "HEALTHY: {{ inputs.target_label }} ({{
              outputs.check_label_population.row.node_count }} nodes) has a
              constraint covering {{ inputs.required_unique_property }}."
        else:
          - id: route_remediation
            type: io.kestra.plugin.core.flow.Switch
            description: auto_remediate only decides whether remediation is considered at
              all; dry_run, checked separately below, decides whether CREATE
              CONSTRAINT actually runs.
            value: "{{ inputs.auto_remediate }}"
            cases:
              "true":
                - id: check_dry_run
                  type: io.kestra.plugin.core.flow.If
                  condition: "{{ inputs.dry_run == \"false\" }}"
                  then:
                    - id: remediate_constraint_gap
                      type: io.kestra.plugin.neo4j.Query
                      description: Creates a uniqueness constraint only if one does not already exist
                        under this generated name - verbatim CREATE CONSTRAINT
                        ... IF NOT EXISTS ... REQUIRE ... IS UNIQUE syntax from
                        Neo4j's own Cypher manual. Schema-only; no node,
                        relationship, or property value is read, created, or
                        deleted.
                      url: "{{ inputs.neo4j_url }}"
                      username: "{{ secret('NEO4J_USERNAME') }}"
                      password: "{{ secret('NEO4J_PASSWORD') }}"
                      query: "CREATE CONSTRAINT {{ inputs.target_label | lower }}_{{
                        inputs.required_unique_property }}_unique IF NOT EXISTS
                        FOR (n:{{ inputs.target_label }}) REQUIRE n.{{
                        inputs.required_unique_property }} IS UNIQUE"
                      storeType: NONE
                    - id: log_remediation_applied
                      type: io.kestra.plugin.core.log.Log
                      message: "Created uniqueness constraint on {{ inputs.target_label }}.{{
                        inputs.required_unique_property }}."
                  else:
                    - id: log_dry_run_remediation
                      type: io.kestra.plugin.core.log.Log
                      message: "DRY RUN: would create a uniqueness constraint on {{
                        inputs.target_label }}.{{
                        inputs.required_unique_property }} (dry_run is true). No
                        schema change made."
              "false":
                - id: log_remediation_disabled
                  type: io.kestra.plugin.core.log.Log
                  message: "auto_remediate is false: {{ inputs.target_label }} is missing a
                    constraint on {{ inputs.required_unique_property }}, but no
                    remediation was attempted."
          - id: alert_constraint_gap
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Always alert on an AT_RISK gap, independent of whether remediation
              ran.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "Neo4j constraint compliance gate: AT_RISK. {{ inputs.target_label }} ({{ outputs.check_label_population.row.node_count }} nodes) has no constraint covering {{ inputs.required_unique_property }} on {{ inputs.neo4j_url }}. auto_remediate: {{ inputs.auto_remediate }}. dry_run: {{ inputs.dry_run }}. Remediated: {{ outputs.remediate_constraint_gap is defined }}. Execution {{ execution.id }}."
              }
    else:
      - id: alert_label_unreachable
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: UNREACHABLE means target_label has no nodes at all in this graph -
          constraint coverage cannot be meaningfully evaluated, so this is
          reported distinctly from both HEALTHY and AT_RISK.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Neo4j constraint compliance gate: UNREACHABLE. {{ inputs.target_label }} has zero nodes on {{ inputs.neo4j_url }} - constraint coverage could not be evaluated. Execution {{ execution.id }}."
          }

  - id: log_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: Final status line for every run, independent of which branch fired,
      so the execution history always shows the headline numbers.
    message: "Constraint audit complete for {{ inputs.target_label }}.{{
      inputs.required_unique_property }} on {{ inputs.neo4j_url }}.
      node_count={{ outputs.check_label_population.row.node_count }}.
      matching_constraints={{
      outputs.check_constraint_coverage.row.matching_constraints }}.
      auto_remediate={{ inputs.auto_remediate }}. dry_run={{ inputs.dry_run }}."

errors:
  - id: alert_execution_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the flow itself fails outside the handled branches, for
      example an unreachable Neo4j instance or invalid credentials.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Neo4j constraint compliance gate FAILED in execution {{ execution.id }} for {{ inputs.target_label }} on {{ inputs.neo4j_url }}. The constraint audit did not complete; investigate before trusting the last known compliance status."
      }