AWS SQS Dead Letter Queue Replay Gate

id: aws-sqs-dead-letter-queue-replay-gate
namespace: company.team
description: |
  Automated Amazon SQS Dead Letter Queue (DLQ) backlog inspector and triage gate. Samples failed messages,
  extracts error attributes, and dispatches actionable Slack alerts before retention timeout expiration.

concurrency:
  limit: 1

inputs:
  - id: dlq_url
    type: STRING
    displayName: SQS Dead Letter Queue URL
    defaults: "https://sqs.us-east-1.amazonaws.com/123456789012/order-processing-dlq"
    description: "Amazon SQS Dead Letter Queue URL to monitor and sample."

  - id: target_queue_url
    type: STRING
    displayName: Primary SQS Target Queue URL
    defaults: "https://sqs.us-east-1.amazonaws.com/123456789012/order-processing-main"
    description: "Primary Amazon SQS queue URL where redriven messages should be routed."

  - id: aws_region
    type: STRING
    displayName: AWS Region
    defaults: "us-east-1"
    description: "AWS region hosting the target SQS queue instances."

  - id: max_messages_to_inspect
    type: INT
    displayName: Inspection Batch Size
    defaults: 10
    description: "Maximum batch of unprocessable messages to sample and analyze per cycle."

  - id: slack_channel
    type: STRING
    displayName: Slack Channel
    defaults: "#ops-alerts"
    description: "Slack channel destination for DLQ backlog escalation alerts."

triggers:
  - id: dlq_check_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Executes DLQ inspection every 30 minutes."
    cron: "*/30 * * * *"

tasks:
  - id: sample_dlq_messages
    type: io.kestra.plugin.aws.sqs.Consume
    description: "Samples a batch of messages from the Dead Letter Queue without
      dropping payloads."
    queueUrl: "{{ inputs.dlq_url }}"
    region: "{{ inputs.aws_region }}"
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    maxRecords: "{{ inputs.max_messages_to_inspect }}"
    visibilityTimeout: 30

  - id: analyze_dead_letter_payloads
    type: io.kestra.plugin.scripts.python.Script
    description: "Inspects failed payload structures, extracts diagnostic headers,
      and compiles triage report."
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    inputFiles:
      consumed.json: "{{ outputs.sample_dlq_messages.messages | default('[]') | json }}"
    outputFiles:
      - dlq_triage_report.json
    script: |
      import json
      from kestra import Kestra

      with open("consumed.json", "r", encoding="utf-8") as f:
          messages = json.load(f)

      count = len(messages)
      triage_records = []

      for idx, msg in enumerate(messages):
          body = msg.get("data", msg.get("body", "{}"))
          attributes = msg.get("attributes", {})
          message_id = msg.get("id", f"msg-{idx}")

          triage_records.append({
              "message_id": message_id,
              "payload_preview": str(body)[:200],
              "attributes": attributes
          })

      report = {
          "dlq_url": "{{ inputs.dlq_url }}",
          "sampled_count": count,
          "has_backlog": count > 0,
          "records": triage_records
      }

      with open("dlq_triage_report.json", "w", encoding="utf-8") as f:
          json.dump(report, f, indent=2)

      Kestra.outputs({
          "backlog_detected": count > 0,
          "message_count": count,
          "sample_preview": triage_records[0]["payload_preview"] if triage_records else "None"
      })

  - id: evaluate_backlog_condition
    type: io.kestra.plugin.core.flow.If
    description: "Branches execution based on whether unprocessable messages were
      detected in the DLQ."
    condition: "{{ outputs.analyze_dead_letter_payloads.vars.backlog_detected }}"
    then:
      - id: alert_on_call_dlq_accumulation
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: "Dispatches operational triage card to on-call engineers via Slack."
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "*AWS SQS Dead Letter Queue Backlog Alert*\n• *DLQ Name*: `{{ inputs.dlq_url }}`\n• *Target Primary Queue*: `{{ inputs.target_queue_url }}`\n• *Sampled Dead Letter Messages*: `{{ outputs.analyze_dead_letter_payloads.vars.message_count }}`\n• *Sample Error Payload Preview*: `{{ outputs.analyze_dead_letter_payloads.vars.sample_preview }}`\n• *Operational Impact*: Failed messages in the DLQ indicate consumer deserialization errors, poison pill payloads, or third-party downstream timeouts.\n• *Recommended Remediation*: Inspect payload schema, resolve downstream consumer issues, and initiate redrive to `{{ inputs.target_queue_url }}`."
          }

    else:
      - id: log_clean_dlq_status
        type: io.kestra.plugin.core.log.Log
        description: "Logs confirmation of empty Dead Letter Queue state."
        message: "AWS SQS DLQ inspection compliant. Zero failed messages detected in {{
          inputs.dlq_url }}."

  - id: export_dlq_manifest
    type: io.kestra.plugin.core.debug.Return
    description: "Outputs structured JSON audit manifest for observability dashboards."
    format: |
      {
        "pipeline": "aws-sqs-dead-letter-queue-replay-gate",
        "audited_at": "{{ execution.startDate }}",
        "dlq_url": "{{ inputs.dlq_url }}",
        "messages_detected": {{ outputs.analyze_dead_letter_payloads.vars.message_count }},
        "status": "{{ outputs.analyze_dead_letter_payloads.vars.backlog_detected ? 'BACKLOG_DETECTED' : 'CLEAN' }}"
      }