Rabbitmq — destinations/rabbitmq

state_stores:
  local:
    namespace: hookway
    memory: {} # Per process; select Redis or PostgreSQL for shared protection.

# examples/destinations/rabbitmq.yaml
#
# Reference config showing the full RabbitMQ destination option surface.
# One flow fans out to two destinations: a primary exchange with dead-letter
# support and an audit exchange with minimal options.

server:
  addr: ":8080"
  read_timeout: "5s"
  write_timeout: "10s"
  idle_timeout: "60s"
  shutdown_timeout: "10s"
  max_body_bytes: 1048576
  # Cap concurrent webhook processing (0 = unlimited).
  max_inflight_requests: 256
  # Cap concurrent fan-out publishes per request (0 defaults to 8).
  fanout_concurrency: 8

destinations:
  primary:
    type: "rabbitmq"
    rabbitmq:
      url: "${RABBITMQ_URL}"
      exchange: "hookway.webhooks"
      exchange_type: "direct"
      durable: true
      publish_timeout: "5s"
      declare_queues: true
      # Publish concurrency: pool of confirm channels. Raise if broker
      # round-trips become the throughput ceiling.
      publisher_channels: 8
      # Retry a nacked confirm or a transient publish/connect failure before
      # giving up. Defaults shown; omit the whole block to use them.
      retry:
        max_attempts: 3
        initial_backoff: "100ms"
        max_backoff: "2s"
      # Default queue arguments applied to all queues this destination declares.
      # There is no per-publish queue_options override.
      queue_defaults:
        # Drop stale deliveries after this much time in queue.
        # RabbitMQ receives this as x-message-ttl in milliseconds.
        message_ttl: "24h"
        # Bound queue growth so a broken consumer cannot fill the broker.
        max_length: 100000
        # Reject new publishes when full so providers retry instead of
        # silently discarding older queued messages.
        overflow: "reject-publish"
      # TLS for amqps:// brokers. Omit entirely for publicly trusted certs
      # (e.g. CloudAMQP) — the container ships the system CA bundle.
      # tls:
      #   ca_cert: "/etc/hookway/ca.pem"
      #   client_cert: "/etc/hookway/tls.crt"
      #   client_key: "/etc/hookway/tls.key"
      #   server_name: "rabbitmq.internal"
      #   insecure_skip_verify: false
      dead_letter:
        enabled: true
        exchange: "hookway.webhooks.dlx"
        exchange_type: "direct"
        queue: "hookway.webhooks.dlq"
        routing_key: "webhooks.dead"
        durable: true
        create_queue: true

  audit:
    type: "rabbitmq"
    rabbitmq:
      url: "${RABBITMQ_URL}"
      exchange: "hookway.audit"
      exchange_type: "direct"
      durable: true
      publish_timeout: "5s"
      declare_queues: true
      publisher_channels: 4

flows:
  paddle_billing:
    # stop_on_failure publishes routes in order and returns 503 on first failure.
    # Use skip_failed when audit/secondary destinations should not block intake.
    delivery_failure_policy: "stop_on_failure"
    nodes:
      - {id: in, type: http_trigger, path: /webhooks/paddle, methods: [POST]}
      - {id: rl, type: rate_limit, store: local, rpm: 500, burst: 1000, key_by: source_ip}
      - {id: body, type: read_body, max_bytes: 1048576}
      - {id: auth, type: auth, scheme: hmac_sha256_ts_colon, secret: "${PADDLE_WEBHOOK_SECRET}", signature_header: Paddle-Signature, timestamp_tolerance: "5m"}
      - {id: guard, type: replay_guard, store: local, ttl: "10m", max_entries: 50000}
      - {id: env, type: envelope, body_format: json}
      # queue_options (primary: message_ttl 72h, max_length 500000,
      # dead_letter_routing_key paddle.dead; audit: message_ttl 168h) are not
      # publish-node fields.
      - {id: out1, type: publish, destination: primary, target: paddle-webhooks, create: true}
      - {id: out2, type: publish, destination: audit, target: paddle-audit-webhooks, create: true}
    edges: [in>rl, rl>body, body>auth, auth>guard, guard>env, env>out1, env>out2]