Audit and Replay Stalled KurrentDB Persistent Subscription Parked Messages

id: kurrentdb-persistent-subscription-parked-message-guard
namespace: company.streaming

description: |
  Audit KurrentDB (formerly EventStoreDB) persistent subscriptions for parked messages,
  the subscription-level dead-letter queue for events that exhausted retries. Track
  backlog persistence across runs in Kestra KV, distinguish a critical surge from a
  slow-building stuck subscription, pause for operator approval, replay parked messages
  back into live processing via the HTTP Admin API, and verify the backlog actually drained.

inputs:
  - id: kurrentdb_url
    type: STRING
    displayName: KurrentDB HTTP Admin API URL
    description: Base HTTP(S) URL of the KurrentDB (EventStoreDB) external HTTP
      port, e.g. https://kurrentdb.internal:2113. Use https unless the cluster
      runs with --insecure.
    defaults: "https://localhost:2113"

  - id: parked_message_threshold
    type: INT
    displayName: Parked Message Threshold
    description: Minimum parkedMessageCount on a single subscription before it is
      tracked as a candidate problem.
    defaults: 1

  - id: critical_parked_threshold
    type: INT
    displayName: Critical Backlog Threshold
    description: parkedMessageCount at or above which a subscription is escalated
      immediately, bypassing the consecutive-run confirmation.
    defaults: 100

  - id: min_consecutive_runs
    type: INT
    displayName: Minimum Consecutive Stuck Runs
    description: Number of consecutive audit passes a sub-critical backlog must
      persist or grow before it is treated as genuinely stuck rather than a
      transient blip.
    defaults: 2

  - id: max_subscriptions_per_run
    type: INT
    displayName: Max Subscriptions Per Run
    description: Pagination cap on the number of persistent subscriptions fetched
      and evaluated in a single audit pass.
    defaults: 50

  - id: replay_stop_at
    type: STRING
    displayName: Replay Stop At Event Number
    description: Optional event number ceiling passed to replayParked so only
      messages parked up to that point are replayed. Leave empty to replay the
      entire parked backlog.
    defaults: ""

  - id: cursor_key
    type: STRING
    displayName: KV Cursor Key
    description: Kestra KV key used to persist per-subscription parked-message trend
      state across runs.
    defaults: "kurrentdb_parked_subscription_tracker"

  - id: dry_run
    type: BOOL
    displayName: Dry Run Mode
    description: When true, audits and classifies subscriptions and logs the
      proposed replayParked calls without mutating any subscription state.
    defaults: true

  - id: auto_remediate
    type: BOOL
    displayName: Auto Remediate
    description: When true and dry_run is false, pauses for operator approval and
      then calls replayParked on verified stalled subscriptions.
    defaults: false

triggers:
  - id: periodic_parked_message_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Audit persistent subscriptions for parked message backlog every 10
      minutes. Shipped disabled so the flow can be verified manually in safe
      mode first.
    cron: "*/10 * * * *"
    disabled: true

  - id: on_demand_parked_message_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated HTTP endpoint invoked by KurrentDB alerting,
      Prometheus Alertmanager, or incident runbooks.
    key: "{{ secret('KURRENTDB_GUARD_WEBHOOK_KEY') }}"

tasks:
  - id: validate_inputs
    type: io.kestra.plugin.core.execution.Assert
    description: Assert operational thresholds satisfy sanity constraints before
      calling the KurrentDB HTTP API.
    conditions:
      - "{{ inputs.parked_message_threshold >= 1 }}"
      - "{{ inputs.critical_parked_threshold >= inputs.parked_message_threshold
        }}"
      - "{{ inputs.min_consecutive_runs >= 1 }}"
      - "{{ inputs.max_subscriptions_per_run >= 1 }}"
    errorMessage: "Invalid configuration: critical_parked_threshold must be >=
      parked_message_threshold, and all thresholds must be positive."

  - id: list_subscriptions
    type: io.kestra.plugin.core.http.Request
    description: List persistent subscriptions registered on the KurrentDB cluster
      via the HTTP Admin API.
    uri: "{{ inputs.kurrentdb_url }}/subscriptions?offset=0&count={{
      inputs.max_subscriptions_per_run }}"
    method: GET
    headers:
      Accept: application/json
    options:
      auth:
        type: BASIC
        username: "{{ secret('KURRENTDB_USERNAME') }}"
        password: "{{ secret('KURRENTDB_PASSWORD') }}"

  - id: triage_subscriptions
    type: io.kestra.plugin.scripts.python.Script
    description: Fetch per-subscription parkedMessageCount, diff against historical
      KV state, and classify stalled subscriptions needing remediation.
    containerImage: python:3.11-slim
    env:
      KURRENTDB_URL: "{{ inputs.kurrentdb_url }}"
      KURRENTDB_USERNAME: "{{ secret('KURRENTDB_USERNAME') }}"
      KURRENTDB_PASSWORD: "{{ secret('KURRENTDB_PASSWORD') }}"
      PARKED_THRESHOLD: "{{ inputs.parked_message_threshold }}"
      CRITICAL_THRESHOLD: "{{ inputs.critical_parked_threshold }}"
      MIN_CONSECUTIVE_RUNS: "{{ inputs.min_consecutive_runs }}"
      EXECUTION_ID: "{{ execution.id }}"
      LIST_SUBSCRIPTIONS_BODY: "{{ outputs.list_subscriptions.body is defined ?
        outputs.list_subscriptions.body : '{}' }}"
      PREVIOUS_STATE_JSON: "{{ kv(inputs.cursor_key, errorOnMissing=false) ?? '{}' }}"
    script: |
      import base64
      import json
      import os
      import urllib.error
      import urllib.request
      from datetime import datetime, timezone

      try:
          from kestra import Kestra
      except ImportError:
          Kestra = None

      now_iso = datetime.now(timezone.utc).isoformat()
      base_url = os.environ.get("KURRENTDB_URL", "").rstrip("/")
      username = os.environ.get("KURRENTDB_USERNAME", "")
      password = os.environ.get("KURRENTDB_PASSWORD", "")
      parked_threshold = int(os.environ.get("PARKED_THRESHOLD", "1"))
      critical_threshold = int(os.environ.get("CRITICAL_THRESHOLD", "100"))
      min_consecutive_runs = int(os.environ.get("MIN_CONSECUTIVE_RUNS", "2"))
      execution_id = os.environ.get("EXECUTION_ID", "manual")
      list_body_raw = os.environ.get("LIST_SUBSCRIPTIONS_BODY", "{}")

      prev_state_raw = os.environ.get("PREVIOUS_STATE_JSON", "{}")
      try:
          prev_state = json.loads(prev_state_raw) if prev_state_raw.strip() else {}
      except Exception:
          prev_state = {}

      auth_header = "Basic " + base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii")

      def http_get_json(path):
          req = urllib.request.Request(
              base_url + path,
              headers={"Authorization": auth_header, "Accept": "application/json"},
              method="GET",
          )
          with urllib.request.urlopen(req, timeout=15) as resp:
              return json.loads(resp.read().decode("utf-8"))

      try:
          list_data = json.loads(list_body_raw) if list_body_raw.strip() else {}
      except Exception:
          list_data = {}

      # GET /subscriptions returns PagedSubscriptionInfo.subscriptions (SubscriptionSummary[]),
      # which does NOT include parkedMessageCount -- only the per-subscription
      # GET /subscriptions/{stream}/{group}/info (SubscriptionInfo) does, so each
      # discovered subscription needs its own follow-up call.
      summaries = list_data.get("subscriptions", []) if isinstance(list_data, dict) else []

      new_state_snapshot = {}
      stalled_subscriptions = []
      healthy_subscriptions = []
      fetch_errors = []

      for summary in summaries:
          if not isinstance(summary, dict):
              continue
          stream_id = str(summary.get("eventStreamId", ""))
          group_name = str(summary.get("groupName", ""))
          if not stream_id or not group_name:
              continue

          key = f"{stream_id}/{group_name}"

          try:
              info = http_get_json(f"/subscriptions/{stream_id}/{group_name}/info")
          except Exception as e:
              fetch_errors.append({"stream": stream_id, "group": group_name, "error": str(e)})
              continue

          parked_count = int(info.get("parkedMessageCount", 0) or 0)
          status = str(info.get("status", "Unknown"))
          in_flight = int(info.get("totalInFlightMessages", 0) or 0)
          outstanding = int(info.get("outstandingMessagesCount", 0) or 0)

          prev_entry = prev_state.get(key, {})
          prev_parked = int(prev_entry.get("parked_message_count", 0))
          prev_consecutive = int(prev_entry.get("consecutive_runs", 0))

          if parked_count >= parked_threshold and parked_count >= prev_parked:
              consecutive_runs = prev_consecutive + 1
          else:
              consecutive_runs = 0

          new_state_snapshot[key] = {
              "parked_message_count": parked_count,
              "consecutive_runs": consecutive_runs,
              "last_evaluated_at": now_iso,
          }

          is_critical = parked_count >= critical_threshold
          is_stuck = parked_count >= parked_threshold and consecutive_runs >= min_consecutive_runs

          if is_critical or is_stuck:
              stalled_subscriptions.append({
                  "event_stream_id": stream_id,
                  "group_name": group_name,
                  "parked_message_count": parked_count,
                  "status": status,
                  "total_in_flight_messages": in_flight,
                  "outstanding_messages_count": outstanding,
                  "consecutive_runs": consecutive_runs,
                  "escalation_reason": "CRITICAL_BACKLOG" if is_critical else "STUCK_CONSECUTIVE_RUNS",
              })
          else:
              healthy_subscriptions.append({
                  "event_stream_id": stream_id,
                  "group_name": group_name,
                  "parked_message_count": parked_count,
              })

      has_stalled = len(stalled_subscriptions) > 0
      overall_status = "CRITICAL" if any(s["escalation_reason"] == "CRITICAL_BACKLOG" for s in stalled_subscriptions) else ("WARNING" if has_stalled else "HEALTHY")

      summary_lines = [
          f"- `{s['event_stream_id']}/{s['group_name']}`: {s['parked_message_count']} parked message(s), "
          f"status={s['status']}, reason={s['escalation_reason']}"
          for s in stalled_subscriptions
      ]

      summary_text = (
          f"KurrentDB Parked Message Guard: {overall_status} | "
          f"Audited {len(summaries)} subscription(s). Stalled: {len(stalled_subscriptions)}. "
          f"Healthy: {len(healthy_subscriptions)}. Fetch errors: {len(fetch_errors)}."
      )
      print(summary_text)

      audit_manifest = {
          "overall_status": overall_status,
          "subscriptions_audited": len(summaries),
          "stalled_count": len(stalled_subscriptions),
          "healthy_count": len(healthy_subscriptions),
          "fetch_error_count": len(fetch_errors),
          "stalled_subscriptions": stalled_subscriptions,
          "execution_id": execution_id,
          "evaluated_at": now_iso,
          "summary": summary_text,
      }

      output_data = {
          "has_stalled_subscriptions": has_stalled,
          "stalled_count": len(stalled_subscriptions),
          "healthy_count": len(healthy_subscriptions),
          "overall_status": overall_status,
          "stalled_subscriptions": stalled_subscriptions,
          "summary": summary_text,
          "summary_details": "\n".join(summary_lines) if summary_lines else "None",
          "new_state_snapshot": new_state_snapshot,
          "audit_manifest": audit_manifest,
      }

      if Kestra:
          Kestra.outputs(output_data)
      else:
          print("::" + json.dumps({"outputs": output_data}) + "::")

  - id: save_tracker_kv
    type: io.kestra.plugin.core.kv.Set
    description: Persist updated per-subscription parked-message trend state into
      Kestra KV for cross-run differential tracking.
    key: "{{ inputs.cursor_key }}"
    value: "{{ outputs.triage_subscriptions.vars.new_state_snapshot | toJson }}"
    kvType: JSON

  - id: evaluate_stalled_subscriptions
    type: io.kestra.plugin.core.flow.If
    description: Branch execution based on whether any persistent subscription is
      flagged with a stalled parked-message backlog.
    condition: "{{ outputs.triage_subscriptions.vars.has_stalled_subscriptions == true }}"
    then:
      - id: dispatch_slack_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Deliver prioritized triage notification to Slack detailing stalled
          subscriptions and their parked backlog.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "attachments": [
              {
                "color": "{{ outputs.triage_subscriptions.vars.overall_status == 'CRITICAL' ? 'danger' : 'warning' }}",
                "title": ":warning: KurrentDB Persistent Subscription Parked Message Alert",
                "text": "*Status:* `{{ outputs.triage_subscriptions.vars.overall_status }}`\n*Cluster:* `{{ inputs.kurrentdb_url }}`\n*Stalled Subscriptions:* {{ outputs.triage_subscriptions.vars.stalled_count }}\n*Mode:* auto_remediate={{ inputs.auto_remediate }}, dry_run={{ inputs.dry_run }}\n*Execution:* `{{ execution.id }}`\n\n*Affected Subscriptions:*\n{{ outputs.triage_subscriptions.vars.summary_details }}\n\n*Summary:* {{ outputs.triage_subscriptions.vars.summary }}"
              }
            ]
          }

      - id: remediation_mode_gate
        type: io.kestra.plugin.core.flow.If
        description: Gate automated remediation behind auto_remediate and dry_run
          configuration flags.
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: pause_for_operator_approval
            type: io.kestra.plugin.core.flow.Pause
            description: Pause workflow execution for operator review before replaying
              parked messages. Resume to apply or kill to abort; no timeout is
              set so this waits indefinitely for a human decision.
            onResume:
              - id: approved
                type: BOOL
                defaults: false
                description: Set true to confirm replayParked should be called against the
                  flagged stalled subscriptions.

          - id: branch_on_approval
            type: io.kestra.plugin.core.flow.If
            description: Execute remediation only if explicit human approval was provided
              during the pause gate.
            condition: "{{ outputs.pause_for_operator_approval.onResume.approved == true }}"
            then:
              - id: remediate_stalled_subscriptions
                type: io.kestra.plugin.core.flow.Loop
                description: Iterate sequentially over stalled subscriptions, replay their
                  parked messages, and verify the backlog drained.
                concurrencyLimit: 1
                values: "{{ outputs.triage_subscriptions.vars.stalled_subscriptions | toJson }}"
                tasks:
                  - id: replay_parked_messages
                    type: io.kestra.plugin.core.http.Request
                    description: Call the KurrentDB replayParked HTTP Admin API endpoint to
                      re-inject parked messages into live subscription
                      processing.
                    uri: "{{ inputs.kurrentdb_url }}/subscriptions/{{
                      fromJson(item.value).event_stream_id }}/{{
                      fromJson(item.value).group_name }}/replayParked{{
                      inputs.replay_stop_at != '' ? ('?stopAt=' ~
                      inputs.replay_stop_at) : '' }}"
                    method: POST
                    options:
                      auth:
                        type: BASIC
                        username: "{{ secret('KURRENTDB_USERNAME') }}"
                        password: "{{ secret('KURRENTDB_PASSWORD') }}"

                  - id: settle_replay
                    type: io.kestra.plugin.core.flow.Sleep
                    description: Allow KurrentDB time to re-dispatch replayed parked messages to
                      live consumers before re-checking the backlog.
                    duration: PT5S

                  - id: verify_parked_drained
                    type: io.kestra.plugin.core.http.Request
                    description: Re-query the subscription's info endpoint to confirm the parked
                      message backlog decreased.
                    uri: "{{ inputs.kurrentdb_url }}/subscriptions/{{
                      fromJson(item.value).event_stream_id }}/{{
                      fromJson(item.value).group_name }}/info"
                    method: GET
                    headers:
                      Accept: application/json
                    options:
                      auth:
                        type: BASIC
                        username: "{{ secret('KURRENTDB_USERNAME') }}"
                        password: "{{ secret('KURRENTDB_PASSWORD') }}"

                  - id: assert_backlog_decreased
                    type: io.kestra.plugin.core.flow.If
                    description: Fail loudly if the parked message count did not decrease after the
                      replay call.
                    condition: "{{ (fromJson(outputs.verify_parked_drained.body).parkedMessageCount
                      ?? 0) >= fromJson(item.value).parked_message_count }}"
                    then:
                      - id: fail_unresolved_backlog
                        type: io.kestra.plugin.core.execution.Fail
                        description: Abort with explicit error details identifying the subscription
                          whose parked backlog did not drain.
                        errorMessage: "replayParked completed but parked message count did not decrease
                          for {{ fromJson(item.value).event_stream_id }}/{{
                          fromJson(item.value).group_name }} (was {{
                          fromJson(item.value).parked_message_count }}, now {{
                          fromJson(outputs.verify_parked_drained.body).parkedMe\
                          ssageCount ?? 'unknown' }}). Check consumer health or
                          inspect the parked stream manually."
                    else:
                      - id: log_subscription_remediated
                        type: io.kestra.plugin.core.log.Log
                        description: Record successful replay and verified backlog reduction for this
                          subscription.
                        message: "Replayed parked messages for {{ fromJson(item.value).event_stream_id
                          }}/{{ fromJson(item.value).group_name }}. Parked
                          count: {{ fromJson(item.value).parked_message_count }}
                          -> {{
                          fromJson(outputs.verify_parked_drained.body).parkedMe\
                          ssageCount ?? 'unknown' }}."

              - id: save_recovery_audit_kv
                type: io.kestra.plugin.core.kv.Set
                description: Persist the remediation audit manifest into Kestra KV for
                  compliance and incident review.
                key: "kurrentdb_parked_audit_{{ execution.id }}"
                value: "{{ outputs.triage_subscriptions.vars.audit_manifest | toJson }}"
                kvType: JSON

              - id: notify_resolution_slack
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                description: Dispatch resolution confirmation to Slack once all flagged
                  subscriptions have been processed.
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                payload: |
                  {
                    "attachments": [
                      {
                        "color": "#2EB67D",
                        "title": ":white_check_mark: KurrentDB Parked Message Backlog Remediated",
                        "text": "*Cluster:* `{{ inputs.kurrentdb_url }}`\n*Subscriptions Processed:* `{{ outputs.triage_subscriptions.vars.stalled_count }}`\n*Execution:* `{{ execution.id }}`"
                      }
                    ]
                  }
            else:
              - id: log_approval_declined
                type: io.kestra.plugin.core.log.Log
                description: Record that the operator declined approval for replaying parked
                  messages.
                message: "Operator declined approval for KurrentDB parked message replay.
                  Stalled subscriptions were left untouched."
        else:
          - id: log_dry_run_plan
            type: io.kestra.plugin.core.log.Log
            description: Log the planned remediation without mutating any subscription
              state.
            message: |
              Dry run active or auto_remediate disabled (dry_run={{ inputs.dry_run }}, auto_remediate={{ inputs.auto_remediate }}).
              Planned replayParked for {{ outputs.triage_subscriptions.vars.stalled_count }} subscription(s):
              {{ outputs.triage_subscriptions.vars.stalled_subscriptions | toJson }}
    else:
      - id: log_healthy_state
        type: io.kestra.plugin.core.log.Log
        description: Confirm nominal state when no persistent subscription has a stalled
          parked-message backlog.
        message: "KurrentDB persistent subscriptions nominal on {{ inputs.kurrentdb_url
          }}. No stalled parked-message backlogs detected."

errors:
  - id: alert_on_workflow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Dispatch emergency alert to Slack if the guard encounters an
      unhandled execution error.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "attachments": [
          {
            "color": "danger",
            "title": ":rotating_light: KurrentDB Parked Message Guard Execution Failed",
            "text": "Failed execution `{{ execution.id }}` while auditing or remediating `{{ inputs.kurrentdb_url }}`. Check cluster connectivity, admin credentials, and worker logs."
          }
        ]
      }

outputs:
  - id: subscriptions_audited
    type: INT
    value: "{{
      outputs.triage_subscriptions.vars.audit_manifest.subscriptions_audited ??
      0 }}"

  - id: stalled_count
    type: INT
    value: "{{ outputs.triage_subscriptions.vars.stalled_count ?? 0 }}"

  - id: status
    type: STRING
    value: "{{ (outputs.triage_subscriptions.vars.stalled_count ?? 0) > 0 ?
      ((inputs.auto_remediate and not inputs.dry_run and
      (outputs.pause_for_operator_approval.onResume.approved ?? false)) ?
      'REMEDIATED' : 'STALLED_SUBSCRIPTIONS') : 'HEALTHY' }}"