Neo4j Index Population Failure Recovery Guard

id: neo4j-index-population-failure-recovery-guard
namespace: company.team

description: |
  Find Neo4j schema indexes left in a non-ONLINE state: FAILED after a population
  error or stuck POPULATING after a crash mid build. Such an index is still listed
  but never serves a lookup, so queries silently fall back to full scans. The flow
  reports them to Slack and can drop then recreate each one from the index own
  definition, wait for population to finish, then re-check that every index is ONLINE.

inputs:
  - id: url
    type: STRING
    displayName: Neo4j Bolt URL
    defaults: "bolt://localhost:7687"

  - id: user
    type: STRING
    displayName: Neo4j user
    defaults: "neo4j"

  - id: password
    type: STRING
    displayName: Neo4j password
    description: Defaults to the NEO4J_PASSWORD secret. Override only for local testing.
    defaults: "{{ secret('NEO4J_PASSWORD') }}"

  - id: await_timeout_seconds
    type: INT
    displayName: Seconds to wait for recreated indexes to come online
    defaults: 120

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, for DROP INDEX or
      CREATE INDEX to run.
    defaults: true

variables:
  audit_cypher: |
    SHOW INDEXES YIELD name, state, type, owningConstraint, failureMessage, createStatement
    WHERE state <> 'ONLINE'
    RETURN name AS index_name, state, type,
           owningConstraint IS NULL AS remediable,
           coalesce(owningConstraint, '') AS owning_constraint,
           left(coalesce(failureMessage, ''), 300) AS failure,
           'DROP INDEX `' + replace(name, '`', '``') + '` IF EXISTS' AS drop_statement,
           createStatement AS create_statement
    ORDER BY index_name

tasks:
  - id: audit
    type: io.kestra.plugin.neo4j.Query
    description: Read-only. SHOW INDEXES is Neo4j own schema metadata, filtered to
      indexes not in the ONLINE state. Each row carries that index
      createStatement, so a recreate uses Neo4j exact definition rather than a
      reconstructed one.
    url: "{{ inputs.url }}"
    username: "{{ inputs.user }}"
    password: "{{ inputs.password }}"
    query: "{{ vars.audit_cypher }}"
    storeType: FETCH

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Neo4j index audit on {{ inputs.url }}: {{ outputs.audit.size }} indexes
      not ONLINE. {% for r in outputs.audit.rows %}{{ r.index_name }} ({{
      r.state }}, {{ r.type }}): {{ r.create_statement }}. {% endfor %}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.audit.size > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Neo4j ' ~ inputs.url ~ ': ' ~ outputs.audit.size ~ ' indexes not ONLINE, still listed but not serving lookups: ' ~ (outputs.audit.rows | jq('map(.index_name + " (" + .state + ")") | join(", ")') | first) ~ '.' ~ (outputs.audit.rows | jq('map(select(.remediable == false)) | length') | first > 0 ? ' Some back a constraint so this guard leaves them alone, rebuild the constraint instead.' : '') ~ (inputs.auto_remediate and not inputs.dry_run ? ' Dropping then recreating now.' : ' No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: remediate
            type: io.kestra.plugin.core.flow.Loop
            concurrencyLimit: 1
            description: One index at a time, remediation of non-ONLINE indexes. Drop then
              recreate from the index own createStatement. A FAILED index keeps
              its name, so CREATE INDEX IF NOT EXISTS is a silent no-op, the
              drop has to run first.
            values: "{{ outputs.audit.rows | toJson }}"
            outputs:
              - id: result
                type: STRING
                value: >-
                  {{ fromJson(item.value).index_name }}: {% if
                  fromJson(item.value).remediable %}drop and recreate issued{%
                  else %}skipped, owned by constraint {{
                  fromJson(item.value).owning_constraint }}{% endif %}
            tasks:
              - id: one
                type: io.kestra.plugin.core.flow.Sequential
                allowFailure: true
                tasks:
                  - id: drop_index
                    type: io.kestra.plugin.neo4j.Query
                    runIf: "{{ fromJson(item.value).remediable }}"
                    url: "{{ inputs.url }}"
                    username: "{{ inputs.user }}"
                    password: "{{ inputs.password }}"
                    query: "{{ fromJson(item.value).drop_statement }}"
                    storeType: NONE

                  - id: recreate_index
                    type: io.kestra.plugin.neo4j.Query
                    runIf: "{{ fromJson(item.value).remediable }}"
                    url: "{{ inputs.url }}"
                    username: "{{ inputs.user }}"
                    password: "{{ inputs.password }}"
                    query: "{{ fromJson(item.value).create_statement }}"
                    storeType: NONE

          - id: settle
            type: io.kestra.plugin.neo4j.Query
            allowFailure: true
            description: Wait for the recreated indexes to finish building. db.awaitIndexes
              throws if one enters FAILED again, which is allowed here. The
              re-audit below is the authoritative check.
            url: "{{ inputs.url }}"
            username: "{{ inputs.user }}"
            password: "{{ inputs.password }}"
            query: "CALL db.awaitIndexes({{ inputs.await_timeout_seconds }})"
            storeType: NONE

          - id: reaudit
            type: io.kestra.plugin.neo4j.Query
            url: "{{ inputs.url }}"
            username: "{{ inputs.user }}"
            password: "{{ inputs.password }}"
            query: "{{ vars.audit_cypher }}"
            storeType: FETCH

          - id: verify_result
            type: io.kestra.plugin.core.flow.If
            description: Fail if any index this run tried to remediate is still not ONLINE.
              Constraint-owned indexes are not counted here, they were never
              touched.
            condition: >-
              {{ {'before': outputs.audit.rows, 'after': (outputs.reaudit.rows
              ?? [])} | jq('([.before[] | select(.remediable == true) |
              .index_name]) as $b | ([.after[].index_name]) as $a | [$b[] |
              select(. as $x | ($a | index($x)) != null)] | length') | first > 0
              }}
            then:
              - id: remediation_failed
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "Indexes still not ONLINE after drop and recreate: {{
                  outputs.reaudit.rows | jq('map(.index_name + \" (\" + .state +
                  \")\") | join(\", \")') | first }}. The cause that failed
                  population is still present, for example a property value too
                  large to index. Fix the data or the definition then re-run."
            else:
              - id: remediated
                type: io.kestra.plugin.core.log.Log
                message: >-
                  {% for r in outputs.remediate.outputs %}{{ r.outputs.result
                  }}. {% endfor %}Indexes not ONLINE left: {{
                  outputs.reaudit.size }}.
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "Every Neo4j index is ONLINE."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Neo4j index population guard failed on ' ~ inputs.url ~ ' (execution ' ~ execution.id ~ '). Index state unknown, check the logs.'} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, always in safe mode.
    cron: "0 4 * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true