Skip to content

/rabbitmq - RabbitMQ integration

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:

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

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:

{
  "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:

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.
  • 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:

{
  "twakeId": "string",
  "iterations": 100000,
  "hash": "base64",
  "publicKey": "base64",
  "privateKey": "cipherString",
  "key": "cipherString",
  "timestamp": 1726040000,
  "domain": "example.cozy.cloud"
}

Example payload for user.created:

{
  "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:

{
  "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:

{
  "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:

{
  "twakeId": "string",
  "mobile": "+33123456789",
  "internalEmail": "string",
  "workplaceFqdn": "example.twake.app"
}

Example message on twake:contacts:common, published by the contacts side service (ADR 043):

{
  "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.

{
  "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):

{
  "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):

{
  "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.

// 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.

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:

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.

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)

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.