RabbitMQ Publisher & Topology

Overview

webhook-go is a producer-only gateway: every endpoint does a small amount of work and hands the rest off to RabbitMQ. There are no consumers, no cron jobs, and no background workers anywhere in this project — all consumers live in worker-go.

Messages are published as bare JSON: the message body is the payload object, with none of the {type, data, meta} envelope that the company template normally applies. The reason is compatibility — the worker-go consumers, ported from NestJS, already parse the raw object, and the code carries an explicit comment warning against changing this.

Routing is as simple as it gets: the exchange is direct, and the routing key always equals the queue name.

Business Flow

Topology declared at boot

cmd/api/main.go calls AMQP.DeclareTopology(...) once the publisher is online. A failure logs a warning and lets the service continue, since the queues may already have been provisioned by the infrastructure team.

1. Declare the main exchange line_exchange as direct and durable.
2. Declare the DLQ exchange line_exchange_dlq as direct and durable.
3. For every queue in the list:
   • assertQueue as durable with the arguments x-max-priority: 10, x-dead-letter-exchange: line_exchange_dlq, and x-dead-letter-routing-key set to the queue name.
   • bindQueue to line_exchange using the queue name as the routing key.
   • assertQueue the paired .dlq queue as durable.
   • bindQueue that .dlq queue to line_exchange_dlq, again keyed by the main queue name.

The declared queues and who actually publishes to them

Queue	Env	Publisher in this project
line_webhook	RABBITMQ_QUEUE_LINE_WEBHOOK	LINE Webhook Gateway, on the default and cache-miss paths
line_forward_webhook	RABBITMQ_QUEUE_LINE_FORWARD_WEBHOOK	None — declared but unused, since forwarding goes over HTTP directly; see Webhook Forward
tracking_log	RABBITMQ_QUEUE_TRACKING_LOG	Tracking Log
mookept_audience	RABBITMQ_QUEUE_MOOKEPT_AUDIENCE_QUEUE	Managing Audience Members, PUT only
mbox_callback	RABBITMQ_QUEUE_MBOX_CALLBACK	MBOX Chatwoot Callback
mbox_handoff	RABBITMQ_QUEUE_MBOX_HANDOFF	MBOX Agent Handoff
booking_notification	RABBITMQ_QUEUE_BOOKING_NOTIFICATION	Booking Cancel Postback

note

.env.example still lists RABBITMQ_QUEUE_LINE_CHANGE_RICHMENU=line_change_richmenu, but internal/config/api.go never reads it — a leftover from the predecessor service.

Publishing via Manager.PublishRaw

1. Marshal the payload to bytes with json.Marshal.
2. Open a fresh channel for each send, so that a broken channel cannot stall the entire publisher.
3. Enable publisher confirms and register the listener before publishing, closing the race window.
4. Send with DeliveryMode: Persistent, ContentType: application/json, and a Priority that is normally 0.
5. Wait for the ack. A missing ack returns an error; an expired context returns ctx.Err().
6. Callers handle failures differently:
   • Fire-and-forget paths (the LINE gateway, tracking) only log the error — the message is lost.
   • Synchronous paths (the mbox callback, PUT /api/audience) propagate the error back to the client.

Connection handling

• AMQP_URLS accepts multiple URLs for HA rotation, handled in internal/amqp/manager.go.
• Publishing is guarded by a mutex because an AMQP channel is not concurrency-safe.
• The publisher is wrapped as a deps.Dependency via internal/deps/amqppub, so it is automatically covered by readiness checks, the supervisor, and graceful shutdown. See Health Checks, Metrics & Dependency Monitoring.
• If AMQP_URLS is empty, deps.AMQP is nil and every publish returns ErrNotConfigured, yet the service still boots and serves health checks.

Key Files & Functions

File	Highlights
internal/amqp/publisher.go	Manager.PublishRaw(ctx, exchange, queue, payload, priority) — the single path every module uses
internal/amqp/topology.go	type Topology and Manager.DeclareTopology(t)
internal/amqp/manager.go	Multi-URL connection manager, Channel(), Ready(), reconnection
internal/amqp/envelope.go	The template's {type,data,meta} envelope — unused in this project
internal/deps/amqppub/amqppub.go	Wraps the publisher as a deps.Dependency (Name/Connect/Ping/Close)
internal/config/api.go	struct RabbitMQ and the Queues() method that fixes declaration order
internal/config/api_load.go	Reads the environment and supplies a default for every queue
cmd/api/main.go	Calls DeclareTopology at boot, guarded by AMQP != nil && Ready()

Connections to Other Services

• worker-go — consumes every queue listed above and is the primary contract partner. Changing a payload shape here is an immediate breaking change on the worker side.
• RabbitMQ cluster — queues and the DLX may be provisioned ahead of time by infrastructure or a CRD; DeclareTopology is idempotent, so re-declaring is safe.
• DLQ — every queue has a .dlq mirror, but this publisher never uses it. Dead-lettering happens when a worker-side consumer rejects a message or exhausts its retries.
• All environment variables are prefixed RABBITMQ_; see .env.example for the full list.