Webhook System

Last Updated: 2026-02-05

FluxMQ can emit broker events to external HTTP endpoints using an asynchronous webhook notifier.

Overview

• Asynchronous event queue with worker pool
• Retry with exponential backoff
• Circuit breaker per endpoint
• Filtering by event type and topic pattern
• HTTP sender only (gRPC sender not implemented)

Architecture

Broker Events
   │
   ▼
Webhook Notifier (queue)
   │  drop_policy: oldest/newest
   ▼
Worker Pool
   │  retry + circuit breaker
   ▼
HTTP Sender
   │
   ▼
External Endpoints

Event Types

Events are defined in broker/events/events.go.

• client.connected: client_id, protocol, clean_start, keep_alive, remote_addr
• client.disconnected: client_id, reason, remote_addr
• client.session_takeover: client_id, from_node, to_node
• message.published: client_id, topic, qos, retained, payload_size, payload
• message.delivered: client_id, topic, qos, payload_size
• message.retained: topic, payload_size, cleared
• subscription.created: client_id, topic_filter, qos, subscription_id
• subscription.removed: client_id, topic_filter
• auth.success: client_id, remote_addr
• auth.failure: client_id, reason, remote_addr
• authz.publish_denied: client_id, topic, reason
• authz.subscribe_denied: client_id, topic_filter, reason

Payload Notes

The message.published payload field is defined in the event schema (base64-encoded when populated), but payload inclusion is not currently wired in the broker. webhook.include_payload is accepted in config, yet payloads are sent as empty strings at the moment.

Event Envelope

{
  "event_type": "message.published",
  "event_id": "550e8400-e29b-41d4-a716-446655440000",
  "timestamp": "2026-02-05T12:00:00Z",
  "broker_id": "broker-1",
  "data": {
    "client_id": "publisher-1",
    "topic": "sensors/temperature",
    "qos": 1,
    "retained": false,
    "payload_size": 256
  }
}

Filtering

Each endpoint can filter by:

• events: list of event types
• topic_filters: MQTT-style patterns (supports + and #)

Retry and Circuit Breaker

• Retries use exponential backoff (initial_interval * multiplier^attempt), capped by max_interval
• Circuit breaker is per endpoint and trips after failure_threshold consecutive failures

Configuration

webhook:
  enabled: true
  queue_size: 10000
  drop_policy: "oldest"          # oldest or newest
  workers: 5
  include_payload: false
  shutdown_timeout: "30s"

  defaults:
    timeout: "5s"
    retry:
      max_attempts: 3
      initial_interval: "1s"
      max_interval: "30s"
      multiplier: 2.0
    circuit_breaker:
      failure_threshold: 5
      reset_timeout: "60s"

  endpoints:
    - name: "analytics"
      type: "http"
      url: "https://example.com/webhook"
      events: ["message.published"]
      topic_filters: ["sensors/#"]
      headers:
        Authorization: "Bearer token"