Solace Dead-Letter Message Triage & Safe Replay Gate

id: solace-dead-letter-message-triage-and-safe-replay-gate
namespace: company.team

description: |
  Drain and triage dead-lettered messages from a Solace PubSub+ queue, classify the batch
  between recoverable processing errors and toxic poison-pill surges, safely re-publish
  valid payloads to the active topic, and isolate catastrophic backlogs to a quarantine
  destination with Slack notifications.

inputs:
  - id: solace_host
    type: STRING
    displayName: Solace Broker Host
    description: Hostname and SMF messaging port for the Solace PubSub+ event broker.
    defaults: "localhost:55555"

  - id: solace_vpn
    type: STRING
    displayName: Solace Message VPN
    description: Message VPN name provisioned on the Solace broker.
    defaults: "default"

  - id: dlq_name
    type: STRING
    displayName: Dead Letter Queue Name
    description: Solace dead-letter queue (DMQ) to drain and audit for failed
      message accumulations.
    defaults: "orders.dlq"

  - id: queue_type
    type: SELECT
    displayName: Queue Access Type
    description: Solace queue durability and access mode as provisioned on the broker.
    values:
      - "DURABLE_NON_EXCLUSIVE"
      - "DURABLE_EXCLUSIVE"
      - "NON_DURABLE_EXCLUSIVE"
    defaults: "DURABLE_NON_EXCLUSIVE"

  - id: target_replay_topic
    type: STRING
    displayName: Target Replay Topic
    description: Active ingestion topic where transiently failed messages are safely
      replayed.
    defaults: "orders/incoming/replayed"

  - id: quarantine_topic
    type: STRING
    displayName: Quarantine Isolation Topic
    description: Topic destination where unrecoverable or surging poison batches are
      isolated.
    defaults: "orders/quarantine/poison"

  - id: max_batch_size
    type: INT
    displayName: Max Messages Per Drain
    description: Maximum number of dead-letter messages to pull in a single triage
      execution.
    defaults: 50

  - id: poll_duration
    type: DURATION
    displayName: Max Drain Duration
    description: Maximum polling time allowed for the consumer task to gather messages.
    defaults: "PT10S"

  - id: backlog_quarantine_ceiling
    type: INT
    displayName: Backlog Quarantine Ceiling
    description: Threshold above which a backlog is classified as an uncontained
      surge and quarantined without auto-replay.
    defaults: 20

triggers:
  - id: solace_dlq_event_trigger
    type: io.kestra.plugin.solace.Trigger
    description: "Event-driven trigger polling the Solace DLQ every 2 minutes.
      Shipped disabled so credentials and destinations can be verified first."
    interval: PT2M
    host: "{{ inputs.solace_host }}"
    username: "{{ secret('SOLACE_USERNAME') }}"
    password: "{{ secret('SOLACE_PASSWORD') }}"
    vpn: "{{ inputs.solace_vpn }}"
    queueName: "{{ inputs.dlq_name }}"
    queueType: "{{ inputs.queue_type }}"
    messageDeserializer: JSON
    maxMessages: 50
    maxDuration: PT10S
    disabled: true

  - id: dlq_triage_schedule_backup
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Safety backup schedule running every 15 minutes to guarantee
      dead-letter audits even if continuous event triggers are paused."
    cron: "*/15 * * * *"
    disabled: true

tasks:
  - id: consume_dead_letter_batch
    type: io.kestra.plugin.solace.Consume
    description: "Drain a bounded batch of dead-lettered messages from the Solace
      queue into Kestra internal storage."
    allowFailure: true
    host: "{{ inputs.solace_host }}"
    username: "{{ secret('SOLACE_USERNAME') }}"
    password: "{{ secret('SOLACE_PASSWORD') }}"
    vpn: "{{ inputs.solace_vpn }}"
    queueName: "{{ inputs.dlq_name }}"
    queueType: "{{ inputs.queue_type }}"
    messageDeserializer: JSON
    maxMessages: "{{ inputs.max_batch_size }}"
    maxDuration: "{{ inputs.poll_duration }}"

  - id: classify_batch_status
    type: io.kestra.plugin.core.debug.Return
    description: "Classify triage state: DRAIN_FAILED if probe unhandled, EMPTY_DLQ
      if 0 messages, EXCESSIVE_POISON_SURGE if exceeding safety ceiling, or
      RECOVERABLE_BATCH."
    format: "{{ outputs.consume_dead_letter_batch.messagesCount is not defined ?
      'DRAIN_FAILED' : (outputs.consume_dead_letter_batch.messagesCount == 0 ?
      'EMPTY_DLQ' : (outputs.consume_dead_letter_batch.messagesCount >
      inputs.backlog_quarantine_ceiling ? 'EXCESSIVE_POISON_SURGE' :
      'RECOVERABLE_BATCH')) }}"

  - id: route_by_triage_decision
    type: io.kestra.plugin.core.flow.Switch
    description: "Branch execution based on the classified dead-letter backlog state."
    value: "{{ outputs.classify_batch_status.value }}"
    cases:
      "RECOVERABLE_BATCH":
        - id: replay_messages_to_active_topic
          type: io.kestra.plugin.solace.Produce
          description: "Safely re-publish drained recoverable messages to the primary
            processing topic with persistent delivery."
          host: "{{ inputs.solace_host }}"
          username: "{{ secret('SOLACE_USERNAME') }}"
          password: "{{ secret('SOLACE_PASSWORD') }}"
          vpn: "{{ inputs.solace_vpn }}"
          topicDestination: "{{ inputs.target_replay_topic }}"
          from: "{{ outputs.consume_dead_letter_batch.uri }}"
          deliveryMode: PERSISTENT
          messageSerializer: JSON

        - id: log_replay_success
          type: io.kestra.plugin.core.log.Log
          message: "Solace DLQ safe replay successful: {{
            outputs.consume_dead_letter_batch.messagesCount }} message(s)
            re-published to topic '{{ inputs.target_replay_topic }}'."

      "EXCESSIVE_POISON_SURGE":
        - id: quarantine_poison_batch
          type: io.kestra.plugin.solace.Produce
          description: "Isolate massive dead-letter backlogs to an offline quarantine
            topic to prevent poison-pill cascade loops."
          host: "{{ inputs.solace_host }}"
          username: "{{ secret('SOLACE_USERNAME') }}"
          password: "{{ secret('SOLACE_PASSWORD') }}"
          vpn: "{{ inputs.solace_vpn }}"
          topicDestination: "{{ inputs.quarantine_topic }}"
          from: "{{ outputs.consume_dead_letter_batch.uri }}"
          deliveryMode: PERSISTENT
          messageSerializer: JSON

        - id: alert_quarantine_surge
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: "Dispatch urgent incident notification to Slack regarding
            suppressed auto-replay and quarantine isolation."
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": "CRITICAL: Solace DLQ '{{ inputs.dlq_name }}' on VPN '{{ inputs.solace_vpn }}' exceeded safety ceiling with {{ outputs.consume_dead_letter_batch.messagesCount }} dead letters. Auto-replay halted to prevent downstream poison cascade. Messages routed to quarantine topic '{{ inputs.quarantine_topic }}'. Execution: {{ execution.id }}."
            }

      "EMPTY_DLQ":
        - id: log_clean_pass
          type: io.kestra.plugin.core.log.Log
          message: "Solace DLQ '{{ inputs.dlq_name }}' on VPN '{{ inputs.solace_vpn }}' is
            clear. Zero failed messages accumulated."

      "DRAIN_FAILED":
        - id: alert_drain_failure
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: "Alert Slack when dead-letter queue cannot be drained due to
            network or authentication failure."
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": "ALERT: Failed to drain Solace DLQ '{{ inputs.dlq_name }}' on host '{{ inputs.solace_host }}' (VPN '{{ inputs.solace_vpn }}'). Broker connection or queue bind failed. Execution: {{ execution.id }}."
            }

    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        message: "Solace DLQ triage finished with unclassified status: {{
          outputs.classify_batch_status.value }}."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: "Immutable audit log recorded on every run documenting queue state,
      message volume, and disposition."
    message: "Solace DLQ Audit: dlq={{ inputs.dlq_name }}, vpn={{ inputs.solace_vpn
      }}, status={{ outputs.classify_batch_status.value }}, count={{
      outputs.consume_dead_letter_batch.messagesCount is defined ?
      outputs.consume_dead_letter_batch.messagesCount : 0 }}, executionId={{
      execution.id }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: "Alert Slack if the gate flow itself encounters an unhandled
      runtime error."
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Solace DLQ Triage Gate flow '{{ flow.id }}' encountered an unhandled execution failure (execution {{ execution.id }}). Triage state unverified."
      }

outputs:
  - id: triage_status
    type: STRING
    value: "{{ outputs.classify_batch_status.value }}"

  - id: messages_drained
    type: INT
    value: "{{ outputs.consume_dead_letter_batch.messagesCount is defined ?
      outputs.consume_dead_letter_batch.messagesCount : 0 }}"