docs/rabbitmq.md
Rama master · réplica de lectura del repositorio original.
## RabbitMQ
Integration with RabbitMQ: configuration, topology, and handler semantics.
### Overview
Cozy Stack can consume messages from RabbitMQ and dispatch them to Go handlers. The consumers are managed by background managers which:
- establish and monitor the AMQP connection with their node (with optional TLS),
- declare exchanges and queues (if configured),
- bind queues to routing keys,
- start per-queue consumers with QoS and redelivery limits,
- dispatch deliveries to queue-specific handlers.
### Configuration
RabbitMQ is configured in `cozy.yaml` under the `rabbitmq` key.
Key fields:
- `enabled`: Enable communication with RabbitMQ nodes.
- `nodes`: Map of RabbitMQ node configuration by context
- `<context>`:
- `url`: AMQP URL, e.g. `amqp://guest:guest@localhost:5672/`.
- `tls`: Optional TLS settings (`root_ca`, `insecure_skip_validation`, `server_name`).
- `exchanges[]`: List of exchanges the stack should consume from.
- `name`: Exchange name.
- `kind`: Exchange type (e.g. `topic`).
- `durable`: Whether the exchange is durable.
- `declare_exchange`: If true, the exchange is declared on startup.
- `dlx_name`, `dlq_name`: Optional defaults for queues under this exchange.
- `queues[]`: List of queues to consume.
- `name`: Queue name.
- `bindings[]`: Routing keys to bind to the exchange.
- `declare`: If true, declare the queue on startup.
- `prefetch`: Per-consumer QoS prefetch.
- `delivery_limit`: x-delivery-limit for quorum queues.
- `declare_dlx`: If true, declare the Dead Letter Exchange (DLX) on startup.
- `declare_dlq`: If true, declare the Dead Letter Queue (DLQ) on startup.
- `dlx_name`, `dlq_name`: Optional overrides per queue.
Example:
```yaml
rabbitmq:
enabled: true
nodes:
default:
enabled: true
url: amqp://guest:guest@localhost:5672/
tls:
# root_ca: /etc/ssl/certs/ca.pem
insecure_skip_validation: false
# server_name: rabbit.internal
exchanges:
- name: auth
kind: topic
durable: true
declare_exchange: false
queues:
- name: stack.user.password.updated
declare: true
declare_dlx: true
declare_dlq: true
dlx_name: stack.auth.dlx
dlq_name: stack.dead.letter.user.password.updated
dl_routing_key: password.updated.dead
prefetch: 8
delivery_limit: 5
bindings:
- user.password.updated
- name: stack.user.created
declare: true
declare_dlx: true
declare_dlq: true
dlx_name: stack.auth.dlx
dlq_name: stack.dead.letter.user.created
dl_routing_key: user.created.dead
prefetch: 8
delivery_limit: 5
bindings:
- user.created
- name: stack.user.phone.updated
declare: true
declare_dlx: true
declare_dlq: true
dlx_name: stack.auth.dlx
dlq_name: stack.dead.letter.user.phone.updated
dl_routing_key: user.phone.updated.dead
prefetch: 8
delivery_limit: 5
bindings:
- user.phone.updated
- name: billing
kind: topic
durable: true
declare_exchange: false
queues:
- name: stack.domain.subscription.changed
declare: true
declare_dlx: true
declare_dlq: true
dlx_name: stack.billing.dlx
dlq_name: stack.dead.letter.domain.subscription.changed
dl_routing_key: domain.subscription.changed.dead
prefetch: 8
delivery_limit: 5
bindings:
- domain.subscription.changed
- name: b2b
kind: topic
durable: true
declare_exchange: false
queues:
- name: stack.user.lifecycle.queue
declare: true
declare_dlx: true
declare_dlq: true
dlx_name: stack.b2b.dlx
dlq_name: stack.dead.letter.user.lifecycle
dl_routing_key: domain.user.deleted.dead
prefetch: 8
delivery_limit: 5
bindings:
- domain.user.deleted
- name: stack.b2b.group.lifecycle
declare: true
declare_dlx: true
declare_dlq: true
dlx_name: stack.b2b.dlx
dlq_name: stack.dead.letter.b2b.group.lifecycle
dl_routing_key: b2b.group.dead
prefetch: 8
delivery_limit: 5
bindings:
- b2b.group.created
- b2b.group.updated
- b2b.group.deleted
- b2b.group.member.added
- b2b.group.member.removed
- name: stack.app.commands.queue
declare: true
declare_dlx: true
declare_dlq: true
dlx_name: stack.b2b.dlx
dlq_name: stack.dead.letter.app.commands
dl_routing_key: app.installation.requested.dead
prefetch: 8
delivery_limit: 5
bindings:
- app.installation.requested
- name: platform
kind: topic
durable: true
declare_exchange: false
queues:
- name: stack.banner.commands
declare: true
declare_dlx: true
declare_dlq: true
dlx_name: stack.platform.dlx
dlq_name: stack.dead.letter.banner.commands
dl_routing_key: banner.commands.dead
prefetch: 8
delivery_limit: 5
bindings:
- banner.materialize
- banner.clear
- name: twake:contacts:common
kind: fanout
durable: true
declare_exchange: false
queues:
- name: stack.contacts.common
declare: true
declare_dlx: true
declare_dlq: true
dlx_name: stack.contacts.dlx
dlq_name: stack.dead.letter.contacts.common
dl_routing_key: contacts.common.dead
prefetch: 8
delivery_limit: 5
bindings:
- ""
```
### Dead Letter Exchange (DLX) and Dead Letter Queue (DLQ)
The RabbitMQ integration supports Dead Letter Exchange (DLX) and Dead Letter Queue (DLQ) functionality for handling failed messages:
- **DLX (Dead Letter Exchange)**: An exchange where messages are sent when they cannot be delivered to their original destination.
- **DLQ (Dead Letter Queue)**: A queue bound to the DLX that receives failed messages for analysis or reprocessing.
#### Configuration
DLX and DLQ can be configured at both exchange and queue levels:
- **Exchange level**: Set `dlx_name` and `dlq_name` in the exchange configuration to provide defaults for all queues under that exchange.
- **Queue level**: Set `dlx_name` and `dlq_name` in the queue configuration to override exchange defaults.
- **Declaration control**: Use `declare_dlx` and `declare_dlq` flags to control whether the DLX and DLQ are automatically declared on startup.
#### Example with DLX/DLQ
```yaml
rabbitmq:
enabled: true
nodes:
default:
enabled: true
url: amqp://guest:guest@localhost:5672/
linagora_default:
enabled: false
url: amqp://guest:guest@localhost:5673/
exchanges:
- name: auth
kind: topic
durable: true
declare_exchange: false
queues:
- name: stack.user.password.updated
declare: true
declare_dlx: true
declare_dlq: true
dlx_name: stack.auth.dlx
dlq_name: stack.dead.letter.user.password.updated
dl_routing_key: password.updated.dead
prefetch: 8
delivery_limit: 5
bindings:
- user.password.updated
- name: stack.user.created
declare: true
declare_dlx: true
declare_dlq: true
dlx_name: stack.auth.dlx
dlq_name: stack.dead.letter.user.created
dl_routing_key: user.created.dead
prefetch: 8
delivery_limit: 5
bindings:
- user.created
```
#### Behavior
- When `declare_dlx: true` and a `dlx_name` is provided, the DLX is declared as a fanout exchange on startup.
- When `declare_dlq: true` and a `dlq_name` is provided, the DLQ is declared and bound to the DLX on startup.
- If queue-level `dlx_name`/`dlq_name` are not specified, exchange-level defaults are used.
- Messages that exceed the `delivery_limit` or are rejected will be sent to the DLX and routed to the DLQ.
### Publishers
The Stack also publishes messages to RabbitMQ. Publishers do not declare any
queue or exchange on the Stack side: the exchange must already exist on the
broker, and a queue must be bound by the consuming service. Publishes use the
AMQP `mandatory` flag, so a publish to an exchange with no matching binding
fails with `PublishReturnedError` and the caller is expected to surface the
failure to the user.
#### `auth` exchange
Routing key: `user.deletion.requested`. Published from
`POST /settings/instance/deletion/force` when a user requests permanent
deletion of their account. The payload is the `UserDeletionRequestedMessage`
struct in `pkg/rabbitmq/contracts.go`.
#### `migration` exchange
Routing key: `nextcloud.migration.requested`. Published from
`POST /remote/nextcloud/migration` when a user starts a Nextcloud-to-Cozy
bulk migration. The payload is the `NextcloudMigrationRequestedMessage`
struct in `pkg/rabbitmq/contracts.go`:
```json
{
"migrationId": "d4e5f6a7b8c94d0ea1b2c3d4e5f6a7b8",
"workplaceFqdn": "alice.cozy.example.com",
"accountId": "a1b2c3d4e5f6",
"sourcePath": "/",
"timestamp": 1712563200
}
```
Credentials are never in the payload: they live in the `io.cozy.accounts`
document referenced by `accountId`. The Stack populates `MessageID` with the
migration ID for cross-system tracing. The consuming service is responsible
for declaring its queue, binding it to this exchange, and processing the
messages; if no queue is bound when the Stack publishes, the user receives
`503` and the tracking document is marked `failed`.
#### `activity` exchange
CloudEvents read by other apps, with the event type as routing key and
`source` `twake://drive`:
- `com.twake.drive.org_drive.ready.v1`: the reply to `drive.org_drive.ensure`,
see below.
- `com.twake.drive.folder.changed.v1`: the activity digest of an organization
drive, see below.
### Handlers
Handlers implement a simple interface:
```go
type Handler interface {
Handle(ctx context.Context, d amqp.Delivery) error
}
```
Returning `nil` acknowledges the message. Returning a non-nil error causes the message to be requeued (subject to broker policies and delivery limits).
A failure is nacked with requeue, and the queue's `delivery_limit` is what
bounds the retries: once it is reached the broker dead letters the message. A
handler can return an error wrapping `rabbitmq.ErrMalformed` for a message no
redelivery can fix: it is nacked without requeue, and goes to the dead letter
queue at once.
Queue names are mapped to handlers in the stack. For example:
- `user.password.updated` → updates an instance passphrase when a `user.password.updated` routing key is received.
- `user.created` → validates and processes user creation events. Writes organization contacts only on the org instance and copies no directory to new members when `centralized_org_directory` is set.
- `user.phone.updated` → updates the phone number stored in user settings.
- `domain.user.deleted` on the `b2b` exchange → removes externally managed organization contacts, except feed-owned contacts while `common_contacts` is enabled.
- `banner.materialize` and `banner.clear` on the `platform` exchange → materializes or clears a platform banner, see [Banners](banners.md).
- every message of the `twake:contacts:common` fanout exchange → writes a contact, see below.
- `drive.org_drive.ensure` on the `drive` exchange → creates or updates a shared drive on an organization instance, see below.
Message schemas are JSON and validated in the handler. Example payload for `user.password.updated`:
```json
{
"twakeId": "string",
"iterations": 100000,
"hash": "base64",
"publicKey": "base64",
"privateKey": "cipherString",
"key": "cipherString",
"timestamp": 1726040000,
"domain": "example.cozy.cloud"
}
```
Example payload for `user.created`:
```json
{
"twakeId": "string",
"mobile": "string",
"internalEmail": "string",
"iterations": 100000,
"hash": "base64",
"publicKey": "base64",
"privateKey": "cipherString",
"key": "cipherString",
"timestamp": 1726040000,
"workplaceFqdn": "alice.example.twake.app",
"organizationId": "org-uuid-123",
"organizationDomain": "example.com",
"canUpgrade": false,
"matrixId": "@alice:example.twake.app"
}
```
`matrixId` is optional. When it is present and well-formed, it is saved as
`matrix_id` in the instance settings document and sent to the common settings
API as the account's Matrix identity; otherwise the Matrix ID is derived from
the email local part and the instance domain.
Example payload for `b2b/domain.user.deleted`:
```json
{
"emitter": "admin-panel",
"type": "user.deleted",
"workplaceFqdn": "alice.example.twake.app",
"internalEmail": "alice@example.com",
"reason": "user deleted",
"organizationId": "org-uuid-123",
"userId": "alice",
"domain": "example.com"
}
```
The delete handler removes the single matching contact found by `internalEmail`
when `metadata.external` is true.
Example payload for `b2b.group.created`:
```json
{
"timestamp": "2026-06-16T10:30:00.000Z",
"organizationId": "org123",
"id": "engineering",
"name": "engineering",
"description": "Engineering team",
"color": "#3366FF",
"createdAt": "2026-06-16T10:30:00.000Z",
"members": []
}
```
The Stack consumes `b2b.group.created`, `b2b.group.updated`,
`b2b.group.deleted`, `b2b.group.member.added`, and
`b2b.group.member.removed` from one lifecycle queue. Groups are replicated as
managed `io.cozy.contacts.groups` documents in every instance with the matching
`org_id`, and `b2b.group.member.added` creates the ones an instance is missing.
Group membership is stored on managed external contacts via
`relationships.groups.data`; existing generated-ID external contacts are reused
by matching email or Cozy URL.
When the context sets `centralized_org_directory`, groups are written on the
organization instance only, independently of `common_contacts`. A member contact
the feed already wrote keeps its fields and only gets the group. Sharing looks
up managed B2B groups on the organization instance even while replication is
retained. Local fallback is allowed only while replication is retained.
Keep `centralized_org_directory: false` for the initial backend rollout. The
current sharing UI still queries local contacts/groups, and group changes on the
organization instance do not yet propagate to sharings on member instances.
Example payload for `user.phone.updated`:
```json
{
"twakeId": "string",
"mobile": "+33123456789",
"internalEmail": "string",
"workplaceFqdn": "example.twake.app"
}
```
Example message on `twake:contacts:common`, published by the contacts side
service ([ADR 043](https://github.com/linagora/twake-workplace-private/pull/1608)):
```json
{
"audience": { "domain": "acme.com" },
"action": "ADD",
"path": "addressbooks/6740b0e4e0c5c1001f2ef9d1/domain-members/a1b2c3.vcf",
"uid": "a1b2c3",
"payload": {
"@type": "Card",
"name": { "full": "Alice Doe" },
"emails": { "e1": { "address": "alice@acme.com" } },
"vCardProps": [["x-twake-workplace-fqdn", {}, "unknown", "alice.twake.app"]]
}
}
```
The feed carries every user and domain. The Stack writes a `domain` audience on
the organization instance whose `org_domain` matches, and a `user` audience on
the instance whose `email` matches. A message for no instance, for
an instance whose context does not set `common_contacts`, or that is malformed
(invalid JSON, no path, no payload, unknown action), is acked and dropped.
The contact is found by its CardDAV `path`, stored as `carddavPath`. A contact
the Stack wrote before the feed knew it (`metadata.external`, like a member
copied by the organization directory) is found by email and takes the path. A
contact typed by hand is never taken over: the feed writes its own beside it. `ADD` and `UPDATE` overwrite the name, emails and phones, and
keep the fields that belong to the Stack (`cozy`, `trustedForSharing`, the
groups). `x-twake-workplace-fqdn` fills `cozy` when it is empty. A message that
changes nothing writes nothing. `DELETE` removes the document.
When a sharing by email creates a contact, and `common_contacts` is set, the
Stack publishes it on the `twake:contacts:collected` fanout exchange. The
request does not wait: it pushes a job for the `share-collected-contacts`
worker, which retries a failed publish. The contacts created by a request go
in a single message, even when the request fails after creating them, since
they stay on the instance. Each one comes back on `twake:contacts:common` and
takes its path. Only contacts created by the sharing are published: a contact
that already exists locally is not backfilled.
```json
{
"userEmail": "alice@acme.com",
"collectedContacts": [
{ "@type": "Card", "emails": { "e1": { "address": "dave@example.org" } } }
]
}
```
#### Organization drives
Another app keeps a shared drive on an organization instance in sync with one
of its objects by sending `drive.org_drive.ensure` on the `drive` exchange
([ADR 009](adr/009-adr-org-drive-provisioning.md)):
```json
{
"organizationId": "org123",
"reference": { "source": "twake-space", "id": "3b9e2c71-5d4a-4f0e-9c8b-1a2d6e7f8091" },
"revision": 1791106503000,
"state": "created",
"name": "Product launch",
"members": [{ "email": "alice@example.net", "role": "write" }],
"groups": [{ "id": "marketing", "role": "read" }],
"activity": true
}
```
- The drive is created on the instance whose slug is `organizationId`, as a
folder at its root named after `name` ("Name (2)" when the name is taken),
and the reference is stored on its sharing.
- Provisioning is locked per organization and reference. Multiple active
drives for one reference fail the command without changing any drive or
publishing a reply; the error lists the conflicting sharing ids.
- A command with a higher revision than the drive's renames it. An older one
is ignored.
- The members and groups of the drive are made to match the lists: a member is
the contact of the organization directory with this email, and a group is
the directory group with this id. Those who are no longer listed are
revoked. Direct member roles use the existing read-only update methods.
Adding members and groups uses the same conflict checks as the sharing API;
conflicting roles are rejected, not combined. Changing the role of an already
attached group is also rejected by the existing model. Each command is applied,
even with the same revision, so a redelivery finishes what a failure left
undone, invitations included.
- A drive whose sharing was revoked or whose folder was trashed is replaced by
a new one. Its old sharing stays inactive and retains the reference and
revision, including after the folder is permanently deleted. An active
sharing whose folder is missing is marked inactive before its replacement
is created.
- After a created command, the Stack replies on the `activity` exchange with
`com.twake.drive.org_drive.ready.v1`, carrying the reference, the revision
and the drive id (`data.drive.id`, the sharing id). `data.skipped` lists the
members (`email`) and groups (`group_id`) that could not be put on the
drive, with a `reason`: not in the organization directory, several contacts
in it, or too many members. A command with the same revision gets a reply
again. Failed invitations fail the delivery before the reply is published;
redelivery retries the unsent invitations. A failed publish also fails the
delivery.
- A command with `"state": "removed"` (no name, members or groups needed)
closes the drives of the reference at once: their routes answer 403. A
background job calls `Sharing.Revoke()`, moves the folder to the trash, and
replies with `com.twake.drive.org_drive.removed.v1` (reference and
revision). The normal trash-retention policy applies, through the context
setting `fs.auto_clean_trashed_after`; without it, there is no automatic purge.
An older revision is ignored, and a reference without a drive still gets the reply.
- A malformed command (no reference, no revision, unknown state or role) goes
to the dead letter queue at once.
- `activity` turns on the activity digest of the drive, below. Each command
sets it again.
#### Drive activity digest
For a drive with `activity` on, whose `ready` reply was confirmed, the Stack
counts the files that members create, update and send to the trash through
`/sharings/drives/:id/`, and publishes the counts at the end of a time window
([ADR 010](adr/010-adr-drive-activity-digest.md)):
```json
{
"specversion": "1.0",
"id": "5c0e9f3a-2b71-5d4e-8a96-7f1c3d2e4b50",
"source": "twake://drive",
"type": "com.twake.drive.folder.changed.v1",
"time": "2026-10-12T09:15:30Z",
"twakeorg": "org123",
"twakeactor": "alice@example.net",
"data": {
"object": {
"type": "folder",
"id": "8ff66e0ee58e",
"title": "Q4 assets",
"container": { "kind": "drive", "id": "a1f0c3e2d4b5" }
},
"files": { "created": 1200, "updated": 3, "deleted": 0 },
"samples": ["logo.svg", "banner.png", "brief.pdf", "budget.xlsx", "notes.md"]
}
}
```
- There is one event per member and top folder (the folder directly under the
root of the drive) that changed in the window. A file at the root counts
under the root folder.
- `container.id` is the drive id of the `ready` reply.
- `samples` holds up to five file names.
- `id` is the same when a flush is retried, so a consumer can drop the events
it has already seen.
- The window is the context setting `drive_activity.window`, 30 seconds by
default. The counts are held in Redis, so a Stack without Redis publishes
no digest.
### Lifecycle
On startup, if `rabbitmq.enabled` is true:
1. The RabbitMQ service starts a manager for every context with configuration.
2. The managers create an AMQP connection with their node (using TLS if configured) and retry with exponential backoff.
3. They declare configured exchanges and queues (if `declare_*` flags are set).
4. They declare Dead Letter Exchanges and Dead Letter Queues (if `declare_dlx`/`declare_dlq` flags are set).
5. They bind queues to their routing keys and start consumers.
6. They expose a readiness channel internally so tests can wait until consumption is active.
7. They monitor the connection and restart consumers upon reconnection.
### Adding a new queue handler
Follow these steps to introduce a new queue and its handler.
1) Define the message schema and the handler
Create a handler type that implements the `Handle(ctx, d)` method and an accompanying message struct.
```go
// Example message payload consumed from the queue
type ExampleEvent struct {
ID string `json:"id"`
Action string `json:"action"`
Timestamp int64 `json:"timestamp"`
}
// ExampleHandler processes ExampleEvent messages
type ExampleHandler struct{}
func NewExampleHandler() *ExampleHandler { return &ExampleHandler{} }
func (h *ExampleHandler) Handle(ctx context.Context, d amqp.Delivery) error {
var msg ExampleEvent
if err := json.Unmarshal(d.Body, &msg); err != nil {
return fmt.Errorf("example: invalid payload: %w", err)
}
if msg.ID == "" {
return fmt.Errorf("example: missing id")
}
// TODO: implement business logic
return nil // ack
}
```
2) Register the handler for a queue name
Map the queue name to the handler so the manager knows which handler to use. This usually happens where queues are built from config.
```go
switch configQueue.Name {
case "example.queue":
handler = NewExampleHandler()
}
```
A package that `pkg/rabbitmq` cannot import registers its handler from `init`
instead. The factory gets the Stack's publisher, for handlers that reply:
```go
func init() {
rabbitmq.RegisterHandler("example.queue", func(p rabbitmq.Publisher) rabbitmq.Handler {
return NewExampleHandler(p)
})
}
```
3) Configure the exchange and queue in `cozy.yaml`
Add the queue under an exchange with the routing keys to bind.
```yaml
rabbitmq:
enabled: true
nodes:
default:
url: amqp://guest:guest@localhost:5672/
exchanges:
- name: auth
kind: topic
durable: true
declare_exchange: true
queues:
- name: example.queue
declare: true
prefetch: 8
delivery_limit: 5
bindings:
- example.created
- example.updated
```
4) Publish a message (example)
```go
ch.PublishWithContext(ctx,
"auth", // exchange
"example.created", // routing key
false, false,
amqp.Publishing{
DeliveryMode: amqp.Persistent,
ContentType: "application/json",
Body: []byte(`{"id":"123","action":"created","timestamp":1726040000}`),
},
)
```
### Independent contacts rollout flags
Both context settings default to false and must be consistent across an organization:
- `common_contacts` enables the common-contacts service (feed and collected contacts).
- `centralized_org_directory` stops copying the directory to member instances.
B2B lifecycle events still maintain the organization instance when the service is off.
Ship organization lookup with `centralized_org_directory: false` first. This keeps
replication for clients that query local contacts and groups. Enable centralization
only after clients support organization lookup and group changes propagate to their
sharings. Enabling the service alone must not stop replication. Changing either
setting does not backfill missing contacts or restore previously deleted copies.