Backfill missing Neo4j relationships a loader never wrote

id: neo4j-missing-relationship-backfill-reconcile
namespace: company.team

description: Reconcile a Neo4j graph against its relational system of record and
  backfill only the relationships a half finished loader never wrote.

concurrency:
  limit: 1

inputs:
  - id: neo4j_url
    type: STRING
    defaults: bolt://neo4j:7687
    description: Neo4j Bolt URI the Kestra worker can reach. Point it at the graph
      that the loader writes.
  - id: neo4j_username
    type: STRING
    defaults: "{{ secret('NEO4J_USERNAME') }}"
    description: Neo4j user. Needs read on the whole graph plus CREATE_RELATIONSHIP
      on the relationship types in scope. No node write right is required.
  - id: neo4j_password
    type: STRING
    defaults: "{{ secret('NEO4J_PASSWORD') }}"
    description: Neo4j password. Prefer the NEO4J_PASSWORD secret.
  - id: source_url
    type: STRING
    defaults: jdbc:postgresql://postgres:5432/app
    description: JDBC URL of the relational system of record that declares which
      edges should exist. The flow adds connection and statement timeouts.
  - id: source_username
    type: STRING
    defaults: "{{ secret('SOURCE_USERNAME') }}"
    description: Read only database user for the authoritative view. SELECT on that
      one view is enough.
  - id: source_password
    type: STRING
    defaults: "{{ secret('SOURCE_PASSWORD') }}"
    description: Database password. Prefer the SOURCE_PASSWORD secret.
  - id: source_view
    type: STRING
    defaults: public.expected_edges
    description: Relation holding one row per edge the graph should contain, with
      columns rel_type, src_label, src_key, dst_label, dst_key. Must be a plain
      lower case identifier, optionally schema qualified.
  - id: rel_types
    type: ARRAY
    itemType: STRING
    defaults: [ PLACED, CONTAINS, IN_CATEGORY, SHIPPED_TO ]
    description: The only relationship types this flow may create. Types left out
      are never written, even when the source declares edges for them. Each name
      must match ^[A-Z][A-Z0-9_]*$.
  - id: max_expected
    type: INT
    defaults: 5000
    min: 1
    max: 20000
    description: Refuse the run when the authoritative view returns more rows than
      this. Keeps one execution inside a single readable plan.
  - id: max_edges
    type: INT
    defaults: 50
    min: 1
    max: 500
    description: Maximum relationships created per approved run, ordered by type
      then source key then destination key. Remaining gaps wait for the next
      run. There is no unlimited mode.
  - id: dry_run
    type: BOOL
    defaults: true
    description: Report the gap and the bounded plan without writing to Neo4j.
  - id: auto_remediate
    type: BOOL
    defaults: false
    description: With dry_run false, request approval before creating the bounded
      set of relationships.
  - id: slack_webhook_url
    type: STRING
    defaults: "{{ secret('SLACK_WEBHOOK_URL') }}"
    description: Slack incoming webhook. Override with a local HTTP sink only in a
      disposable test.

variables:
  jdbc_url: "{{ inputs.source_url }}{{ inputs.source_url contains '?' ? '&' : '?'
    }}connectTimeout=5&socketTimeout=40&options=-c%20statement_timeout%3D30000"
  source_sql: |
    WITH scoped AS MATERIALIZED (
      SELECT rel_type, src_label, src_key, dst_label, dst_key
      FROM {{ inputs.source_view }}
      WHERE rel_type IN (SELECT value FROM jsonb_array_elements_text(CAST(:rel_types_json AS jsonb)))
      ORDER BY rel_type, src_key, dst_key
      LIMIT CAST(:max_expected AS bigint) + 1
    ), checked AS MATERIALIZED (
      SELECT *, (
        rel_type ~ '^[A-Z][A-Z0-9_]{0,63}$'
        AND src_label ~ '^[A-Za-z][A-Za-z0-9_]{0,63}$'
        AND dst_label ~ '^[A-Za-z][A-Za-z0-9_]{0,63}$'
        AND src_key ~ ('^[A-Za-z0-9][A-Za-z0-9_.' || chr(58) || '-]{0,127}$')
        AND dst_key ~ ('^[A-Za-z0-9][A-Za-z0-9_.' || chr(58) || '-]{0,127}$')
      ) AS safe
      FROM scoped
    )
    SELECT count(*) AS expected_total,
      count(*) FILTER (WHERE NOT safe) AS unsafe_total,
      CAST(coalesce(jsonb_agg(jsonb_build_array(rel_type, src_label, src_key, dst_label, dst_key)
        ORDER BY rel_type, src_key, dst_key) FILTER (WHERE safe), CAST('[]' AS jsonb)) AS text) AS pairs_json
    FROM checked
  diff_tail: |
    WITH {rel_type: t[0], src_label: t[1], src_key: t[2], dst_label: t[3], dst_key: t[4]} AS row
    OPTIONAL MATCH (s {key: row.src_key}) WHERE row.src_label IN labels(s)
    WITH row, s
    OPTIONAL MATCH (d {key: row.dst_key}) WHERE row.dst_label IN labels(d)
    WITH row, s, d
    OPTIONAL MATCH (s)-[r]->(d) WHERE type(r) = row.rel_type
    WITH row, s IS NOT NULL AS src_exists, d IS NOT NULL AS dst_exists, count(r) AS present
    RETURN row.rel_type AS rel_type, row.src_label AS src_label, row.src_key AS src_key,
      row.dst_label AS dst_label, row.dst_key AS dst_key,
      src_exists, dst_exists, present,
      (present = 0 AND src_exists AND dst_exists) AS writable
    ORDER BY rel_type, src_key, dst_key
  census_cypher: |
    MATCH (n)
    RETURN 'graph' AS kind, 'nodes' AS name, count(n) AS total, 0 AS startNodes, 0 AS endNodes
    UNION ALL
    MATCH (n)
    UNWIND labels(n) AS lbl
    WITH lbl AS name, count(*) AS total
    RETURN 'label' AS kind, name, total, 0 AS startNodes, 0 AS endNodes
    UNION ALL
    MATCH (a)-[r]->(b)
    WITH type(r) AS name, count(r) AS total, count(DISTINCT a) AS startNodes, count(DISTINCT b) AS endNodes
    RETURN 'relType' AS kind, name, total, startNodes, endNodes

triggers:
  - id: hourly_graph_reconcile
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "25 * * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false

tasks:
  - id: safe_scope
    type: io.kestra.plugin.core.execution.Assert
    conditions:
      - "{{ (inputs.rel_types | length) > 0 and (inputs.rel_types | length) <=
        20 }}"
      - "{{ (inputs.rel_types | toJson | jq('[.[] |
        select(test(\"^[A-Z][A-Z0-9_]{0,63}$\"))] | length') | first) ==
        (inputs.rel_types | length) }}"
      - "{{ ([inputs.source_view] | toJson | jq('[.[] |
        select(test(\"^[a-z_][a-z0-9_]{0,62}([.][a-z_][a-z0-9_]{0,62})?$\"))] |
        length') | first) == 1 }}"
    errorMessage: Give between 1 and 20 upper case relationship type names and a
      plain lower case relation identifier for source_view. Anything else is
      refused before Neo4j is touched.

  - id: read_source
    type: io.kestra.plugin.jdbc.postgresql.Query
    connectionPooling: false
    url: "{{ render(vars.jdbc_url) }}"
    username: "{{ inputs.source_username }}"
    password: "{{ inputs.source_password }}"
    fetchType: FETCH_ONE
    parameters:
      rel_types_json: "{{ inputs.rel_types | toJson }}"
      max_expected: "{{ inputs.max_expected }}"
    sql: "{{ render(vars.source_sql) }}"

  - id: trusted_source
    type: io.kestra.plugin.core.execution.Assert
    conditions:
      - "{{ outputs.read_source.row.unsafe_total == 0 }}"
      - "{{ outputs.read_source.row.expected_total > 0 and
        outputs.read_source.row.expected_total <= inputs.max_expected }}"
    errorMessage: Refused the authoritative view. It returned no rows at all or more
      rows than max_expected or labels and keys outside the allowed character
      set. Fix or narrow the view before any edge is created.

  - id: detect
    type: io.kestra.plugin.neo4j.Query
    url: "{{ inputs.neo4j_url }}"
    username: "{{ inputs.neo4j_username }}"
    password: "{{ inputs.neo4j_password }}"
    storeType: FETCH
    query: |
      UNWIND {{ outputs.read_source.row.pairs_json }} AS t
      {{ render(vars.diff_tail) }}

  - id: census_detect
    type: io.kestra.plugin.neo4j.Query
    url: "{{ inputs.neo4j_url }}"
    username: "{{ inputs.neo4j_username }}"
    password: "{{ inputs.neo4j_password }}"
    storeType: FETCH
    query: "{{ render(vars.census_cypher) }}"

  - id: plan
    type: io.kestra.plugin.core.output.OutputValues
    values:
      selected_json: "{{ {'rows': outputs.detect.rows, 'limit': inputs.max_edges} |
        toJson | jq('.limit as $l | [.rows[] | select(.writable == true)] |
        sort_by(.rel_type, .src_key, .dst_key) | .[:($l|tonumber)]') | first |
        toJson }}"
      blocked_json: "{{ outputs.detect.rows | toJson | jq('[.[] | select(.present == 0
        and .writable != true) | {rel_type, src_key, dst_key, src_exists,
        dst_exists}]') | first | toJson }}"
      gap_by_type_json: "{{ outputs.detect.rows | toJson | jq('group_by(.rel_type) |
        [.[] | {rel_type: .[0].rel_type, expected: length, present: ([.[] |
        select(.present > 0)] | length), missing: ([.[] | select(.present == 0)]
        | length)}]') | first | toJson }}"
      missing_total: "{{ outputs.detect.rows | toJson | jq('[.[] | select(.present ==
        0)] | length') | first }}"
      writable_total: "{{ outputs.detect.rows | toJson | jq('[.[] | select(.writable
        == true)] | length') | first }}"

  - id: targets
    type: io.kestra.plugin.core.output.OutputValues
    values:
      selected_total: "{{ outputs.plan.values.selected_json | jq('length') | first }}"
      approved_pairs_json: "{{ outputs.plan.values.selected_json | jq('[.[] |
        [.rel_type, .src_label, .src_key, .dst_label, .dst_key]]') | first |
        toJson }}"
      expected_add_json: "{{ outputs.plan.values.selected_json |
        jq('group_by(.rel_type) | map({key: .[0].rel_type, value: length}) |
        from_entries') | first | toJson }}"

  - id: report
    type: io.kestra.plugin.core.log.Log
    message: >-
      Authoritative edges in scope: {{ outputs.read_source.row.expected_total
      }}. Missing in Neo4j: {{ outputs.plan.values.missing_total }}, of which {{
      outputs.plan.values.writable_total }} have both endpoint nodes already in
      the graph. Gap per type: {{ outputs.plan.values.gap_by_type_json }}.
      Bounded plan of {{ outputs.targets.values.selected_total }} relationships:
      {{ outputs.plan.values.selected_json }}. Refused because an endpoint node
      does not exist, so the node loader has to run first: {{
      outputs.plan.values.blocked_json }}. Graph census before any change: {{
      outputs.census_detect.rows | toJson }}. Relationship types outside
      rel_types and edges the view does not declare are reported nowhere here
      and are never touched.

  - id: findings
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (outputs.plan.values.missing_total | number) > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ inputs.slack_webhook_url }}"
        payload: "{{ {'text': 'Neo4j graph reconcile found ' ~
          outputs.plan.values.missing_total ~ ' declared relationships missing,
          ' ~ outputs.targets.values.selected_total ~ ' of them writable now.
          Traversals that cross those edges return nothing. Read the bounded
          plan in execution ' ~ execution.id ~ '. dry_run=' ~ inputs.dry_run ~ '
          auto_remediate=' ~ inputs.auto_remediate} | toJson }}"
      - id: mode
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run and
          (outputs.targets.values.selected_total | number) > 0 }}"
        then:
          - id: approval
            type: io.kestra.plugin.core.flow.Pause
            pauseDuration: PT1H
            behavior: FAIL
            description: Approve the exact plan in the report task. Each item runs one MERGE
              between two nodes that already exist. Nodes are matched and never
              created. Nothing is deleted. Confirm the node loader finished and
              that no other loader owns these relationship types.
            onResume:
              - id: approved
                type: BOOL
                required: false
                description: Set true to approve. Omitted or false means Neo4j is left
                  untouched.
          - id: approved_only
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approval.onResume.approved ?? false }}"
            then:
              - id: recheck
                type: io.kestra.plugin.neo4j.Query
                url: "{{ inputs.neo4j_url }}"
                username: "{{ inputs.neo4j_username }}"
                password: "{{ inputs.neo4j_password }}"
                storeType: FETCH
                query: |
                  UNWIND {{ outputs.targets.values.approved_pairs_json }} AS t
                  {{ render(vars.diff_tail) }}
              - id: fresh_plan
                type: io.kestra.plugin.core.execution.Assert
                conditions:
                  - "{{ outputs.recheck.size ==
                    (outputs.targets.values.selected_total | number) }}"
                  - "{{ (outputs.recheck.rows | toJson | jq('[.[] |
                    select(.writable != true)] | length') | first) == 0 }}"
                errorMessage: The graph moved while the plan waited for approval. At least one
                  approved relationship now exists or lost an endpoint node.
                  Nothing was written. Run detection again and approve the new
                  plan.
              - id: census_before
                type: io.kestra.plugin.neo4j.Query
                url: "{{ inputs.neo4j_url }}"
                username: "{{ inputs.neo4j_username }}"
                password: "{{ inputs.neo4j_password }}"
                storeType: FETCH
                query: "{{ render(vars.census_cypher) }}"
              - id: backfill
                type: io.kestra.plugin.core.flow.Loop
                concurrencyLimit: 1
                values: "{{ outputs.plan.values.selected_json }}"
                tasks:
                  - id: merge_one
                    type: io.kestra.plugin.neo4j.Query
                    allowFailure: true
                    url: "{{ inputs.neo4j_url }}"
                    username: "{{ inputs.neo4j_username }}"
                    password: "{{ inputs.neo4j_password }}"
                    storeType: FETCH
                    query: |
                      MATCH (s {key: '{{ fromJson(item.value).src_key }}'}) WHERE '{{ fromJson(item.value).src_label }}' IN labels(s)
                      MATCH (d {key: '{{ fromJson(item.value).dst_key }}'}) WHERE '{{ fromJson(item.value).dst_label }}' IN labels(d)
                      MERGE (s)-[r:{{ fromJson(item.value).rel_type }}]->(d)
                      RETURN type(r) AS rel_type, startNode(r).key AS src_key, endNode(r).key AS dst_key
              - id: verify
                type: io.kestra.plugin.neo4j.Query
                url: "{{ inputs.neo4j_url }}"
                username: "{{ inputs.neo4j_username }}"
                password: "{{ inputs.neo4j_password }}"
                storeType: FETCH
                query: |
                  UNWIND {{ outputs.targets.values.approved_pairs_json }} AS t
                  WITH {rel_type: t[0], src_label: t[1], src_key: t[2], dst_label: t[3], dst_key: t[4]} AS row
                  OPTIONAL MATCH (s {key: row.src_key}) WHERE row.src_label IN labels(s)
                  WITH row, s
                  OPTIONAL MATCH (d {key: row.dst_key}) WHERE row.dst_label IN labels(d)
                  WITH row, s, d
                  OPTIONAL MATCH (s)-[r]->(d) WHERE type(r) = row.rel_type
                  WITH row, count(r) AS present
                  RETURN row.rel_type AS rel_type, row.src_key AS src_key, row.dst_key AS dst_key,
                    present, (present = 1) AS landed
                  ORDER BY rel_type, src_key, dst_key
              - id: census_after
                type: io.kestra.plugin.neo4j.Query
                url: "{{ inputs.neo4j_url }}"
                username: "{{ inputs.neo4j_username }}"
                password: "{{ inputs.neo4j_password }}"
                storeType: FETCH
                query: "{{ render(vars.census_cypher) }}"
              - id: invariants
                type: io.kestra.plugin.core.output.OutputValues
                values:
                  drift_json: >-
                    {{ {'before': outputs.census_before.rows, 'after':
                    outputs.census_after.rows, 'add':
                    fromJson(outputs.targets.values.expected_add_json)} | toJson
                    | jq('(.before | map({key: (.kind + "|" + .name), value:
                    .total}) | from_entries) as $b | (.after | map({key: (.kind
                    + "|" + .name), value: .total}) | from_entries) as $a |
                    (.add) as $add | ([($b | keys), ($a | keys)] | add | unique)
                    as $keys | [$keys[] | . as $k | ($k | split("|")) as $p |
                    {kind: $p[0], name: $p[1], before: ($b[$k] // 0), after:
                    ($a[$k] // 0), expected_add: (if $p[0] == "relType" then
                    ($add[$p[1]] // 0) else 0 end)} | . + {ok: (.after ==
                    (.before + .expected_add))}]') | first | toJson }}
              - id: write_outcome
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Read back from Neo4j after the backfill, one row per approved
                  relationship, landed means present exactly once: {{
                  outputs.verify.rows | toJson }}. Census drift against the
                  counts taken immediately before the first write: {{
                  outputs.invariants.values.drift_json }}. Rows with landed
                  false were refused by Neo4j or lost an endpoint node. They
                  stay in the gap and the next run replans them.
              - id: verdict
                type: io.kestra.plugin.core.execution.Assert
                conditions:
                  - "{{ (outputs.verify.rows | toJson | jq('[.[] |
                    select(.landed != true)] | length') | first) == 0 }}"
                  - "{{ (outputs.invariants.values.drift_json | jq('[.[] |
                    select(.ok != true)] | length') | first) == 0 }}"
                errorMessage: An approved relationship did not land exactly once or the graph
                  changed beyond the approved edges. Read the write_outcome
                  report for the per relationship read back and the census
                  drift, then run detection again. Relationships that did land
                  are kept and the next run plans only what is still missing.
              - id: reconciled
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Created {{ outputs.targets.values.selected_total }} declared
                  relationships, each verified present exactly once: {{
                  outputs.verify.rows | toJson }}. Node counts, out of scope
                  relationship types and undeclared edges are unchanged: {{
                  outputs.invariants.values.drift_json }}. Traversals that cross
                  these edges now return results. Remaining gaps, if any, wait
                  for the next run.
            else:
              - id: declined
                type: io.kestra.plugin.core.log.Log
                message: Approval declined or omitted. Neo4j was not written to and the gap
                  report stands.
        else:
          - id: report_only
            type: io.kestra.plugin.core.log.Log
            message: Dry run, remediation disabled or nothing writable yet because the
              missing edges still lack an endpoint node. Neo4j was not written
              to.
    else:
      - id: healthy
        type: io.kestra.plugin.core.log.Log
        message: Every relationship the authoritative view declares for the types in
          scope is already present in the right direction. Nothing to backfill.

errors:
  - id: failure_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ inputs.slack_webhook_url }}"
    payload: "{{ {'text': 'Neo4j graph reconcile failed in execution ' ~
      execution.id ~ '. Check whether the plan went stale during approval or a
      MERGE was refused. Relationships already created are kept and the next run
      replans from the live graph.'} | toJson }}"