Per-org AMQP Queues

Overview

Voke models each integrating organisation as an isolated AMQP tenant. A single RabbitMQ topic exchange named vcp carries every VCP message. Each partner API key gets its own dedicated vhost named partner-{apiKeyId} — federation mirrors the relevant routing keys from the central exchange into that vhost so the partner only ever sees their own org's traffic. Isolation is enforced by two layers: the per-key vhost itself, and the RabbitMQ HTTP auth backend, which validates every connect, queue, and publish operation against the authenticated key and its org.

Queue layout

All queues are prefixed with vcp.{queuePrefix} and declared durable: true (survive broker restarts). For most orgs queuePrefix is the org slug; if the operator overrode the prefix at provisioning time (the voke org has slug=voke but queuePrefix=default, for example), use that override everywhere instead. The exact prefix to use for your org is in the connection bundle returned at mint time — read it from connectionBundle.queues.* rather than reconstructing names yourself.

Inbound = you publish to the exchange; Voke's consumer picks it up. Outbound = Voke publishes; you consume.

Routing key conventions

Inbound

Commands, config, and schedules use a {slug}.{family}.{subtype} pattern:

{slug}.command.site-setpoint    — site-level power / energy setpoint
{slug}.command.device           — per-device command (charge/discharge/stop)
{slug}.command.emergency        — emergency stop or hold

{slug}.config.constraints       — operating constraints (max power, SOC limits, etc.)
{slug}.config.topology          — asset declaration and topology update
{slug}.config.fallback          — fallback behaviour on loss of connectivity

{slug}.schedule.create          — create or replace a dispatch schedule
{slug}.schedule.cancel          — cancel an active schedule

The command.# binding on vcp.{slug}.command and config.# binding on vcp.{slug}.config use multi-word wildcards, so additional command or config types can be added without modifying the binding. The schedule.* binding uses a single-word wildcard — schedule subtypes are always one token.

Outbound

Voke publishes outbound events using the following routing keys (verified from VcpAmqpService):

The alarm and execution bindings use the multi-word wildcard (#) so the current bare routes and any future subtype routes both land in the same queue. Status uses single-word subtype bindings (event.command.*, event.mode.*, event.schedule.*).

AMQP auth flow

When a partner's AMQP client connects to RabbitMQ, the broker has no local user database. Instead, it delegates every auth decision to Voke via the rabbitmq_auth_backend_http plugin. Voke exposes four endpoints under /api/v1/internal/amqp-auth/. These endpoints are @Public() (no JWT required) and intentionally carry no application-layer header secret — the rabbitmq_auth_backend_http plugin (v3.13.7) does not support sending custom headers, so an earlier X-Internal-Secret design was dropped. Protection comes from two layers below the app: only the broker container can reach the API on the internal Docker network, and the nginx reverse proxy returns 404 for any /api/v1/internal/* request from the public internet.

Endpoint details

POST /api/v1/internal/amqp-auth/user

Called once on connect. Voke:

1. Hashes the password using HMAC-SHA256 (with pepper) and looks up the API key.
2. Checks that the key is active and not expired.
3. Validates that the API key's organisation slug matches the supplied username.
4. Checks that the key has the vcp:connect scope.

Returns "allow" on success, "deny" on any failure.

POST /api/v1/internal/amqp-auth/vhost

Called after user auth to check vhost access.

Allowed vhosts:

• / — the default vhost. Returned by apiKey.vhost === null for keys that weren't provisioned with a per-key vhost (the deployment-wide PARTNER_PER_KEY_VHOST_ENABLED flag is off, or the key carries no vcp:* scopes).
• partner-{apiKeyId} — dedicated per-key vhost when the key was provisioned with a vhost value.

Any other vhost is denied. For per-key vhosts, the API key must be active, its stored apiKey.vhost must match the requested vhost, and the key's org slug must match the AMQP username.

POST /api/v1/internal/amqp-auth/resource

Called per operation (queue declare, bind, publish, consume). Voke enforces:

• Exchange access: vcp topic exchange → allowed (this is the only one partners need). The default exchange ("") is also allowed for AMQP-level direct-reply RPC.
• Queue access: only queues whose name starts with vcp.{orgSlug}. are allowed. Any attempt to touch another org's queue returns "deny".
• All other resource types → "deny".

POST /api/v1/internal/amqp-auth/topic

Called per publish/consume on topic exchanges. Voke enforces that the routing key starts with the org's queue prefix. A partner for org acme cannot publish to beta.command.site-setpoint or consume from beta.event.telemetry.realtime.

Write permission is also scope-gated:

Unknown publish routing-key families fail closed. Read/consume operations remain prefix-gated.

Queue lifecycle

Queues are managed dynamically — Voke never requires a manual broker setup.

On enable: When an org's trading status transitions to ENABLED, TradingPartnerConfigService.enableTrading() calls VcpAmqpService.addOrgConsumer(orgId, slug). OrgVcpConsumer.setup() then asserts all seven queues (durable: true) and binds them to the vcp exchange. If AMQP is not yet connected, the consumer is queued and set up as soon as the connection is established.

On disable: disableTrading() calls VcpAmqpService.removeOrgConsumer(orgId), which calls OrgVcpConsumer.teardown(deleteQueues = false). By default, queues are not deleted — they are unbound from consumers and their contents preserved. Pass deleteQueues = true explicitly to delete them (e.g. during integration teardown).

On API boot: VcpAmqpService.onModuleInit() connects to AMQP and calls setupAllOrgConsumers(), which queries all TradingPartnerConfig rows with status = ENABLED and re-asserts queues for each. PerVhostConsumerService also opens one consumer connection for each active API key with a non-null vhost. Queues survive API restarts without any operator action.

On AMQP disconnect: The service clears all in-process OrgVcpConsumer instances and schedules a reconnect with exponential backoff (initial 1 s, max 30 s). On reconnect all org consumers are re-established automatically.

Durability and ack guarantees

All queues are declared durable: true. Messages published by Voke use persistent: true (delivery mode 2). Messages survive a RabbitMQ broker restart for all queues.

For inbound queues (you publish):

• Publish with persistent: true (delivery mode 2) to ensure messages survive a broker restart before Voke picks them up.
• Wrap each publish in a channel confirm (channel.waitForConfirms()) if you need delivery guarantees.

For outbound queues (you consume):

• Acknowledge (ack) after successful processing.
• On malformed payloads (schema validation failure), nack with requeue = false so the message routes to a dead-letter queue rather than looping.
• Voke's own inbound consumers follow this pattern: parse and validate with Zod; ack on success; nack(requeue=false) on parse failure.

Connection endpoint

Partner clients connect to the public AMQPS listener — amqps:// over TLS, never plaintext.

The host and port that Voke embeds in the partner connection bundle are read from PUBLIC_AMQPS_HOST (default amqp.voke.turena.cz) and PUBLIC_AMQPS_PORT (default 5671). If you're running a local Voke stack, set those env vars on the API so the bundle points at your machine; the API ships no separate "localhost" default. The separately-named AMQP_PUBLIC_HOST / AMQP_PUBLIC_PORT env vars are unrelated — they configure the API's own Docker-network connection to the broker and are not the partner-facing surface.

Connection string pattern:

// username = org slug, password = plaintext API key secret (vcp:connect scope)
// vhost = apiKey.vhost (per-key, the common case) or '/' (when per-key vhost isn't provisioned)
const vhostPath = apiKey.vhost ? `/${encodeURIComponent(apiKey.vhost)}` : '/';
const url = `amqps://${encodeURIComponent(orgSlug)}:${encodeURIComponent(apiKeySecret)}@${host}:${port}${vhostPath}`;

The Connect Partner wizard's reveal screen hands you the fully-assembled amqps:// URI from ConnectionBundleBuilder — paste that directly rather than reconstructing it. See API keys & auth for how to create a key and which scope to assign.