Purge expired Neo4j sessions in approved bounded batches

id: neo4j-expired-node-ttl-purge-guard
namespace: company.team
description: Purge approved expired Neo4j session and event nodes in bounded
  transactions while preserving nodes with other labels or non-disposable
  relationships.
concurrency:
  limit: 1
  behavior: QUEUE
inputs:
  - id: neo4j_url
    type: STRING
    defaults: "{{ secret('NEO4J_URL') }}"
    description: Bolt endpoint reachable by the worker.
  - id: neo4j_username
    type: STRING
    defaults: neo4j
    description: User with complete graph visibility.
  - id: neo4j_password
    type: SECRET
    defaults: "{{ secret('NEO4J_PASSWORD') }}"
    description: Database credential.
  - id: retention_labels
    type: STRING
    defaults: Session,Event
    description: Comma-separated exact labels. Every label on a deletable node must
      be listed.
    validator: ^[^,\r\n]+(,[^,\r\n]+)*$
  - id: expiry_property
    type: STRING
    defaults: expires_at
    description: Integer Unix seconds. Escaped into the query as a string literal.
    validator: ^[^\r\n]{1,128}$
  - id: disposable_relationship_types
    type: STRING
    defaults: NEXT_EVENT
    description: Comma-separated exact relationship types whose edges may be
      deleted. Every other type protects the node.
    validator: ^[^,\r\n]+(,[^,\r\n]+)*$
  - id: cutoff_epoch
    type: INT
    defaults: 0
    description: Strict expiry cutoff in Unix seconds. Zero uses database time.
      Future cutoffs are refused.
  - id: max_nodes
    type: INT
    defaults: 500
    description: Hard run budget. Refuse the entire plan when exceeded.
    min: 1
    max: 1000
  - id: batch_size
    type: INT
    defaults: 50
    description: Maximum nodes per transaction.
    min: 1
    max: 100
  - id: max_degree
    type: INT
    defaults: 10
    description: Maximum incident relationships per eligible node.
    min: 0
    max: 20
  - id: auto_remediate
    type: BOOL
    defaults: false
    description: Permit the approval path.
  - id: dry_run
    type: BOOL
    defaults: true
    description: Must be false before approval or writes.
  - id: slack_webhook_url
    type: STRING
    defaults: "{{ secret('SLACK_WEBHOOK_URL') }}"
    description: Webhook for execution failures.
variables:
  audit_cypher: |-
    UNWIND split('{{ inputs.retention_labels | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ',') AS label
    OPTIONAL MATCH (n) WHERE label IN labels(n)
    RETURN label, count(n) AS nodes,
           count(CASE WHEN valueType(n['{{ inputs.expiry_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}']) = 'INTEGER NOT NULL' AND n['{{ inputs.expiry_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}'] < toInteger({{ outputs.clock.row.cutoff }}) THEN 1 END) AS expired,
           count(CASE WHEN valueType(n['{{ inputs.expiry_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}']) = 'INTEGER NOT NULL' AND n['{{ inputs.expiry_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}'] < toInteger({{ outputs.clock.row.cutoff }})
    AND size(labels(n)) > 0
    AND all(l IN labels(n) WHERE l IN split('{{ inputs.retention_labels | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ','))
    AND COUNT { (n)--() } <= toInteger({{ inputs.max_degree }})
    AND NOT EXISTS { MATCH (n)-[r]-() WHERE NOT type(r) IN split('{{ inputs.disposable_relationship_types | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ',') } THEN 1 END) AS eligible
    ORDER BY label
  plan_cypher: |-
    MATCH (n)
    WHERE valueType(n['{{ inputs.expiry_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}']) = 'INTEGER NOT NULL' AND n['{{ inputs.expiry_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}'] < toInteger({{ outputs.clock.row.cutoff }})
    AND size(labels(n)) > 0
    AND all(l IN labels(n) WHERE l IN split('{{ inputs.retention_labels | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ','))
    AND COUNT { (n)--() } <= toInteger({{ inputs.max_degree }})
    AND NOT EXISTS { MATCH (n)-[r]-() WHERE NOT type(r) IN split('{{ inputs.disposable_relationship_types | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ',') }
    WITH n ORDER BY elementId(n) LIMIT toInteger({{ inputs.max_nodes }}) + 1
    CALL {
      WITH n
      OPTIONAL MATCH (n)-[r]-()
      WITH DISTINCT elementId(r) AS rid ORDER BY rid
      RETURN collect(rid) AS rels
    }
    WITH elementId(n) + '|' + toString(n['{{ inputs.expiry_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}']) + '|' +
         reduce(s = '', r IN rels | s + CASE WHEN s = '' THEN '' ELSE ',' END + r) AS token
    ORDER BY token
    WITH collect(token) AS tokens
    RETURN tokens,
           reduce(s = '', t IN tokens | s + CASE WHEN s = '' THEN '' ELSE ',' END + split(t, '|')[0]) AS ids,
           [i IN range(0, size(tokens) - 1, toInteger({{ inputs.batch_size }})) |
             reduce(s = '', t IN tokens[i..i + toInteger({{ inputs.batch_size }})] |
               s + CASE WHEN s = '' THEN '' ELSE '
    ' END + t)] AS batches
  census_cypher: |-
    CALL {
      MATCH (n) UNWIND labels(n) AS label
      WITH label, count(*) AS total WHERE NOT label IN split('{{ inputs.retention_labels | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ',')
      RETURN 'label' AS kind, label AS name, total
      UNION ALL
      MATCH ()-[r]->() WITH type(r) AS name, count(*) AS total
      WHERE NOT name IN split('{{ inputs.disposable_relationship_types | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ',')
      RETURN 'relationship' AS kind, name, total
    }
    RETURN kind, name, total ORDER BY kind, name
tasks:
  - id: clock
    type: io.kestra.plugin.neo4j.Query
    url: '{{ inputs.neo4j_url }}'
    username: '{{ inputs.neo4j_username }}'
    password: '{{ inputs.neo4j_password }}'
    timeout: PT1M
    storeType: FETCHONE
    query: |-
      WITH timestamp() / 1000 AS current, toInteger({{ inputs.cutoff_epoch }}) AS requested
      RETURN CASE WHEN requested = 0 THEN current ELSE requested END AS cutoff,
             requested >= 0 AND requested <= current AS valid
  - id: valid_cutoff
    type: io.kestra.plugin.core.execution.Fail
    condition: '{{ not outputs.clock.row.valid }}'
    errorMessage: Use zero for database time or a non-negative past cutoff in Unix seconds.
  - id: audit
    type: io.kestra.plugin.neo4j.Query
    url: '{{ inputs.neo4j_url }}'
    username: '{{ inputs.neo4j_username }}'
    password: '{{ inputs.neo4j_password }}'
    timeout: PT1M
    storeType: FETCH
    query: '{{ render(vars.audit_cypher) }}'
  - id: plan
    type: io.kestra.plugin.neo4j.Query
    url: '{{ inputs.neo4j_url }}'
    username: '{{ inputs.neo4j_username }}'
    password: '{{ inputs.neo4j_password }}'
    timeout: PT1M
    storeType: FETCHONE
    query: '{{ render(vars.plan_cypher) }}'
  - id: retention_report
    type: io.kestra.plugin.core.log.Log
    message: At cutoff {{ outputs.clock.row.cutoff }} the retention census is {{
      outputs.audit.rows | toJson }}. Planned {{ outputs.plan.row.tokens |
      length }} nodes; other expired nodes remain protected by labels,
      relationships or degree.
  - id: within_budget
    type: io.kestra.plugin.core.execution.Fail
    condition: '{{ (outputs.plan.row.tokens | length) > inputs.max_nodes }}'
    errorMessage: Expired-node plan exceeds max_nodes. Nothing was deleted. Review
      the retention scope or raise the budget within its hard ceiling.
  - id: purge_gate
    type: io.kestra.plugin.core.flow.If
    condition: '{{ inputs.auto_remediate and not inputs.dry_run and
      (outputs.plan.row.tokens | length) > 0 }}'
    then:
      - id: approval
        type: io.kestra.plugin.core.flow.Pause
        description: Permanently delete the planned expired nodes and their disposable
          edges. Confirm the policy and backup before approving.
        pauseDuration: PT20M
        behavior: FAIL
        onResume:
          - id: approved
            type: BOOL
            defaults: false
      - id: approved_purge
        type: io.kestra.plugin.core.flow.If
        condition: '{{ outputs.approval.onResume.approved ?? false }}'
        then:
          - id: fresh_plan
            type: io.kestra.plugin.neo4j.Query
            url: '{{ inputs.neo4j_url }}'
            username: '{{ inputs.neo4j_username }}'
            password: '{{ inputs.neo4j_password }}'
            timeout: PT1M
            storeType: FETCHONE
            query: '{{ render(vars.plan_cypher) }}'
          - id: unchanged_plan
            type: io.kestra.plugin.core.execution.Fail
            condition: '{{ (outputs.fresh_plan.row.tokens | toJson) !=
              (outputs.plan.row.tokens | toJson) }}'
            errorMessage: The approved node, expiry or edge set changed. Nothing was
              deleted. Run detection again.
          - id: protected_before
            type: io.kestra.plugin.neo4j.Query
            url: '{{ inputs.neo4j_url }}'
            username: '{{ inputs.neo4j_username }}'
            password: '{{ inputs.neo4j_password }}'
            timeout: PT1M
            storeType: FETCH
            query: '{{ render(vars.census_cypher) }}'
          - id: batches
            type: io.kestra.plugin.core.flow.Loop
            concurrencyLimit: 1
            values: '{{ outputs.plan.row.batches | toJson }}'
            tasks:
              - id: delete_batch
                type: io.kestra.plugin.neo4j.Query
                url: '{{ inputs.neo4j_url }}'
                username: '{{ inputs.neo4j_username }}'
                password: '{{ inputs.neo4j_password }}'
                timeout: PT1M
                storeType: FETCHONE
                query: |-
                  UNWIND split('{{ item.value | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', '
                  ') AS token
                  WITH split(token, '|') AS t
                  MATCH (n) WHERE elementId(n) = t[0]
                  WITH n, t ORDER BY elementId(n)
                  SET n['{{ inputs.expiry_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}'] = n['{{ inputs.expiry_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}']
                  WITH n, t
                  WHERE valueType(n['{{ inputs.expiry_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}']) = 'INTEGER NOT NULL' AND n['{{ inputs.expiry_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}'] < toInteger({{ outputs.clock.row.cutoff }})
                  AND size(labels(n)) > 0
                  AND all(l IN labels(n) WHERE l IN split('{{ inputs.retention_labels | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ','))
                  AND COUNT { (n)--() } <= toInteger({{ inputs.max_degree }})
                  AND NOT EXISTS { MATCH (n)-[r]-() WHERE NOT type(r) IN split('{{ inputs.disposable_relationship_types | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ',') } AND n['{{ inputs.expiry_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}'] = toInteger(t[1])
                  CALL {
                    WITH n
                    OPTIONAL MATCH (n)-[r]-()
                    WITH DISTINCT r ORDER BY elementId(r)
                    RETURN collect(r) AS rels
                  }
                  WITH n, t, rels
                  WHERE all(r IN rels WHERE elementId(r) IN CASE WHEN t[2] = '' THEN [] ELSE split(t[2], ',') END)
                  WITH collect({node: n, rels: rels}) AS selected
                  WITH selected, reduce(rs = [], x IN selected | rs + x.rels) AS rels,
                       size(selected) = size(split('{{ item.value | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', '
                  ')) AS complete
                  FOREACH (r IN CASE WHEN complete THEN rels ELSE [] END | DELETE r)
                  FOREACH (n IN CASE WHEN complete THEN [x IN selected | x.node] ELSE [] END | DELETE n)
                  RETURN CASE WHEN complete THEN size(selected) ELSE 0 END AS deleted,
                         NOT complete AS stale_batch
                description: Lock node expiry values then match the approved expiry and reject
                  edges outside the approved set. A stale batch deletes nothing.
                  Plain DELETE refuses an unexpected attached edge.
          - id: remaining
            type: io.kestra.plugin.neo4j.Query
            url: '{{ inputs.neo4j_url }}'
            username: '{{ inputs.neo4j_username }}'
            password: '{{ inputs.neo4j_password }}'
            timeout: PT1M
            storeType: FETCHONE
            query: '{{ render(vars.plan_cypher) }}'
          - id: approved_remaining
            type: io.kestra.plugin.neo4j.Query
            url: '{{ inputs.neo4j_url }}'
            username: '{{ inputs.neo4j_username }}'
            password: '{{ inputs.neo4j_password }}'
            timeout: PT1M
            storeType: FETCHONE
            query: 'UNWIND split(''{{ outputs.plan.row.ids | replace({''\\'': ''\\\\''},
              regexp=true) | replace({"''": "\''"}) }}'', '','') AS id OPTIONAL
              MATCH (n) WHERE elementId(n) = id RETURN count(n) AS remaining'
          - id: protected_after
            type: io.kestra.plugin.neo4j.Query
            url: '{{ inputs.neo4j_url }}'
            username: '{{ inputs.neo4j_username }}'
            password: '{{ inputs.neo4j_password }}'
            timeout: PT1M
            storeType: FETCH
            query: '{{ render(vars.census_cypher) }}'
          - id: verified
            type: io.kestra.plugin.core.execution.Fail
            condition: '{{ (outputs.remaining.row.tokens | length) != 0 or
              outputs.approved_remaining.row.remaining != 0 or
              (outputs.protected_before.rows | toJson) !=
              (outputs.protected_after.rows | toJson) }}'
            errorMessage: Eligible expired nodes remain or protected label/relationship
              counts changed. Completed batches remain committed. Inspect the
              graph and run detection again.
          - id: after
            type: io.kestra.plugin.neo4j.Query
            url: '{{ inputs.neo4j_url }}'
            username: '{{ inputs.neo4j_username }}'
            password: '{{ inputs.neo4j_password }}'
            timeout: PT1M
            storeType: FETCH
            query: '{{ render(vars.audit_cypher) }}'
          - id: purged
            type: io.kestra.plugin.core.log.Log
            message: 'Eligible expired nodes cleared at cutoff {{ outputs.clock.row.cutoff
              }}. Before {{ outputs.audit.rows | toJson }}; after {{
              outputs.after.rows | toJson }}. Protected census unchanged: {{
              outputs.protected_after.rows | toJson }}.'
        else:
          - id: declined
            type: io.kestra.plugin.core.log.Log
            message: Approval declined or omitted. No retention write ran.
    else:
      - id: report_only
        type: io.kestra.plugin.core.log.Log
        message: Read-only run or no eligible expired nodes. No retention write ran.
errors:
  - id: failure_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: '{{ inputs.slack_webhook_url }}'
    payload: '{{ {''text'': ''Neo4j retention purge failed: execution '' ~
      execution.id ~ ''. Inspect stale plans, protected state and remaining
      expired nodes before retrying. Completed batches may remain committed.''}
      | toJson }}'
triggers:
  - id: nightly_retention
    type: io.kestra.plugin.core.trigger.Schedule
    cron: 0 2 * * *
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true