Repository navigation
Appwrite Native Push Broker - #13372
Conversation
…ement Introduce an injected onConnect authenticator so the broker stays free of any Appwrite dependency (extractable to utopia-php/mqtt). CONNECT resolves an identity via the authenticator and rejects with CONNACK 0x87 on failure; the AUTH packet re-verifies on reauth and enforces the projectId matches the one pinned at CONNECT, so a connection cannot switch tenants. Fold the CONNECT and AUTH property blocks into one reader, and name the response/reason-code bytes as constants instead of raw hex.
The MQTT analog of app/init/realtime/connection.php, without the HTTP surface (origin/CSRF, cookies, dev keys, impersonation). Resolves the project from the CONNECT projectId and verifies the session secret via Store::decode + User::sessionVerify — the same transport-agnostic domain logic HTTP uses.
Mirror the realtime process bootstrap (init, DB pool helpers, coroutine runtime) and register the container-backed authenticator on the broker. The authenticator builds a child container from the CONNECT projectId and session secret, resolves project and user, and returns the identity (or empty to reject).
getConsoleDB() returns a Database with its own Authorization instance, so $authorization->skip() was toggling a different object than the one the read checks against — the permission-filtered projects lookup came back empty. Bind the same Authorization before skipping, matching the user resolver.
Thread the CONNECT Authentication Method through the authenticator seam (projectId, authMethod, credential) so the resolver can pick how to read the credential. On "appwrite-jwt" the user factory decodes the HS256 JWT with _APP_OPENSSL_KEY_V1 and checks the session is still active; otherwise it Store-decodes the session blob and runs sessionVerify. Reauth passes the method through the same way. Rename the connection credential from sessionSecret.
…and add blocked user connection rejection test
…e API Collapse the per-packet handlers, Dispatcher and Response into a single Appwrite\Mqtt\Handler that implements the library Handler interface; the Server now owns framing, dispatch, encoding, keep-alive and the subscription store. Resolve broker resources (cache, console/project DB, redis, plan) through the container instead of global functions. Honour each topic's expiry as the replay-cursor TTL (PUBACK preserves it with a 0-ttl save rather than shrinking it to the default hour), and clamp delivery QoS to the subscriber's granted QoS. Drop the adapter unit test whose registry/store behaviour now lives in the library.
| }, ['register']); | ||
| } | ||
|
|
||
| $container->set('getCache', fn () => function () use ($register): Cache { |
There was a problem hiding this comment.
got the suggestion that we need them per coroutine context thats why not doing DI directly rather injecting them in the child container
There was a problem hiding this comment.
Because of the same reason as othere. This is per coroutine and we are saving in the coroutine context
# Conflicts: # composer.lock
| * | ||
| * @param iterable<array{0: Connection, 1: int}> $subscribers | ||
| */ | ||
| public function onPublish(Publish $publish, Connection $connection, iterable $subscribers): void |
There was a problem hiding this comment.
not needed the publish as its getting done internally via the workers here
There was a problem hiding this comment.
maybe it'd be more elegant to give it a optional implemenation then?
There was a problem hiding this comment.
Not optional cause
No optional method possible in the php interfaces. Then we have to either create a default onPublish method or create a separate PublishHandler to have the onPublish. Feeling a bit confusing
publish is a valid spec in mqtt and should be added but we are blocking it and taking no action to block for user. Messaging worker is doing the publish via the pubsub.
…thorizer The DI refactor moved the broker's DB/cache resolvers onto the container but left the authorizer closure calling the removed global getProjectDB(), so every SUBSCRIBE threw "Call to undefined function getProjectDB()" before the SUBACK (CONNECT was unaffected as it only uses the authenticator).
# Conflicts: # composer.lock
Wire a request Authorization onto the broker's project/console DB handles and skip on that same instance: the DB enforces permissions against the instance set via setAuthorization, and topics can carry empty permissions, so the previous skip on a stale/unwired Authorization silently filtered every topic and denied all grants. onSubscribe now establishes the Authorization after the authorizer runs (it mutates the shared handle). Align the SUBACK denial reason code assertions to the library's 0x80. Removes the temporary diagnostics.
Add a WebSocket listener (:8083) alongside raw TCP on the broker, and route ws(s)://<host>/push to it through Traefik (TLS terminated at the edge, mirroring mqtts and realtime's wss). Covers it with an e2e that speaks MQTT 5.0 over a WebSocket binary frame end to end: enhanced-auth CONNECT and an ACL-checked SUBSCRIBE granted at the topic's QoS.
…o PUBACK) A subscription to a QoS-0 topic (grant capped from the client's QoS-1 request) is delivered at QoS 0 and the client sends no PUBACK. Expose the received QoS from the subscriber to lock the QoS-0/QoS-1 segregation in place.
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Drop the plaintext port-80 ws router; the CONNECT frame carries a reusable JWT/session credential, so route /push only through the websecure entrypoint (wss), matching the TLS-only mqtts TCP router. Addresses the credentials-over-plaintext review finding.
…t a /push path Give the WebSocket carrier its own TLS entrypoint (:8084, mqtt-wss) alongside the mqtts TCP entrypoint (:8883), each a catch-all on its port. The transport is now identified by host+port rather than a URL path, so one push endpoint (e.g. push.cloud.appwrite.io) serves both mqtts:// and wss:// and it no longer shares :443/paths with the API or realtime. Still TLS-only.
The WS transport is addressed by its dedicated TLS port, not a URL path, so the client upgrades at / rather than /push.
The dev override replaces Traefik's command wholesale; it only re-declared the mqtts :8883 entrypoint, so the new mqtt-wss :8084 entrypoint was missing and wss was unreachable in the dev stack (the :8084 port publish survives via ports concatenation but had no entrypoint behind it). Re-declare it alongside mqtts.
|
|
||
| go(function () use ($server, $mqtt, $register): void { | ||
| $attempts = 0; | ||
| while ($attempts < 300) { |
There was a problem hiding this comment.
Why 300 attempts? That seems like a crazy number
There was a problem hiding this comment.
Its taken from the realtime. Reducing it a bit. Will adjust this in future after few testings around
| echo "=== Appwrite Logs ===" | ||
| docker compose logs --timestamps | ||
|
|
||
| e2e_mqtt: |
There was a problem hiding this comment.
possible to make it just an additional entry in the service matrix?
| use Utopia\DI\Container; | ||
| use Utopia\System\System; | ||
|
|
||
| return function (Container $container): void { |
There was a problem hiding this comment.
these run per connection, but I don't see the connection actually being used to vary the result? why not make them singletons?
There was a problem hiding this comment.
The result does vary but per coroutine, not per argument. That variation is hidden inside Coroutine::getContext(), not in the signature
If we have a singleton then the coroutines would start sharing them can break tenancy similar to what happened in the realtime which lead to adding of the coroutine context.
getProjectDB($project) does vary by argument
getPlanForUser does make sense to be in the connection
| }, ['register']); | ||
| } | ||
|
|
||
| $container->set('getCache', fn () => function () use ($register): Cache { |
| public function send(string $projectId, array $payload, array $events, array $channels, array $roles, array $options = []): void | ||
| { | ||
| $message = $options['payload'] ?? ''; | ||
| $qos = $options['qos'] ?? 0; | ||
| $sequence = (int) ($options['sequence'] ?? 0); | ||
|
|
||
| $this->messageSize->record(\strlen($message)); | ||
|
|
||
| foreach ($channels as $topic) { | ||
| $this->messagesPublished->add(1, ['qos' => $qos]); | ||
| $this->pubsub->publish(self::CHANNEL, (string) json_encode([ | ||
| 'project' => $projectId, | ||
| 'topic' => $topic, | ||
| 'qos' => $qos, | ||
| 'sequence' => $sequence, | ||
| 'payload' => base64_encode($message), | ||
| ])); | ||
| } | ||
| } |
There was a problem hiding this comment.
not a blocker
does feel like something of this could live in utopia mqtt, pubsub/state adapter (redis/inmemory) + sender
| /** | ||
| * Part of the messaging-adapter contract but unused for MQTT: subscriptions are owned by the | ||
| * broker (utopia-php/mqtt Server records them from the SUBACK the Handler returns). | ||
| * | ||
| * @param array<int, string> $roles | ||
| * @param array<int, string> $channels | ||
| * @param array<int, mixed> $queryGroup | ||
| */ | ||
| public function subscribe(string $projectId, mixed $identifier, string $subscriptionId, array $roles, array $channels, array $queryGroup = []): void | ||
| { | ||
| } | ||
|
|
||
| public function unsubscribe(mixed $identifier): void | ||
| { | ||
| } |
There was a problem hiding this comment.
this is a big signal that the design is trying to fit square in a round role, empty method overrides.
There was a problem hiding this comment.
This is basically a message publisher. And There is nothing to subscribe unlike the FCM/APNS adapter here and just getting used for publishing in the messaging worker which follows a strict interface
| use Utopia\Span\Span; | ||
| use Utopia\System\System; | ||
|
|
||
| class Handler implements MqttHandler |
There was a problem hiding this comment.
This class looks so much better 👍
To go futher, the next step would be, are any parts are common to all MQTT server impelemenations? if they are move to utopia
| * | ||
| * @param iterable<array{0: Connection, 1: int}> $subscribers | ||
| */ | ||
| public function onPublish(Publish $publish, Connection $connection, iterable $subscribers): void |
There was a problem hiding this comment.
maybe it'd be more elegant to give it a optional implemenation then?
There was a problem hiding this comment.
this di is a little bit awkward tbh
I'd aim for a structure that allows you to get the actually things you need
cache and planForUser
directly as parametesrs
The handler no longer service-locates getCache/getPlanForUser off the container by string key; both arrive as typed constructor params. The cache stays a resolver closure because the pool-backed cache can only be built inside a coroutine, so it is resolved lazily per coroutine as before.
The counter resets on a healthy ping, so this caps consecutive failures at ~2 minutes before the delivery bridge gives up.
Drop the standalone e2e_mqtt job; add Mqtt to e2e_service with processes: 1 so the session/replay tests keep running serially against the one broker.
2 participants
MQTT push broker (QoS 1 with offline replay)
Summary
This adds an MQTT 5.0 push broker to Appwrite as a self hosted alternative to FCM. A device connects to the broker over MQTT, authenticates with an Appwrite JWT or session, subscribes to a topic, and receives push messages.
Publishing stays a server privilege. Messages are sent through the existing Messaging campaign API, never over MQTT from a client. There is no client PUBLISH handler.
Delivery is QoS 1 with offline replay. If a device is offline when a message is published, the broker stores the message and replays it on the next reconnect, up to a bounded window.
Where the code lives
app/mqtt.phpis the broker process. It runs a Swoole server on port 1883 (internal only). It parses each packet, dispatches it to a handler, runs the keep-alive reaper, and listens on the pub/sub channel for fan out.app/init/mqtt/connection.phpholds connection auth. It registersauthenticator(turns a CONNECT into an identity) andauthorizer(the per subscribe block check).src/Appwrite/Mqtt/Handlers/has one class per packet type: Connect, Auth, Subscribe, Unsubscribe, Puback, Ping, Disconnect.src/Appwrite/Mqtt/Connection.phpis the per connection state, keyed by file descriptor.src/Appwrite/Mqtt/SubscriptionStore.phpis the topic trie that matches subscribers to a topic.src/Appwrite/Mqtt/KeepAlive.phpis the timing wheel that reaps silent connections.src/Appwrite/Messaging/Adapter/Mqtt.phpis the broker adapter. It holds the connection registry and fans a published message out to local subscribers.src/Appwrite/Messaging/Adapter/Push/Appwrite.phpis the push provider. It writes each message to the ledger and hands it to the broker.Connect and subscribe flow
sequenceDiagram participant D as Device (MQTT client) participant B as Broker (app/mqtt.php) participant DB as Project database participant R as Redis (cursor cache) participant W as Messaging worker D->>B: CONNECT (auth method, credential, projectId) B->>DB: resolve project and user, verify credential DB-->>B: user (or empty) B-->>D: CONNACK (0x00 accepted, 0x87 rejected) D->>B: SUBSCRIBE (topic filter, requested QoS) B->>DB: topic exists and its subscribe roles allow the user B-->>D: SUBACK (granted QoS per filter, or 0x87) B->>R: read the client cursor for the topic B->>DB: read ledger rows after the cursor (capped at maxDepth) B-->>D: PUBLISH x N (offline replay, DUP set) Note over W,B: later, a campaign is published W->>DB: append to ledger, increment topic sequence W->>B: fan out on the mqtt pub/sub channel B-->>D: PUBLISH (live, QoS 1) D->>B: PUBACK B->>R: advance the cursorHow to publish and subscribe
Publish is a normal server side campaign. It targets a topic and routes through the Appwrite push provider.
Subscribe is an MQTT 5 client. It connects over TLS to the Traefik mqtts entrypoint on 8883 and passes the credential through MQTT 5 enhanced auth. The topic filter is the topic id.
The credential is a JWT (
authentication-method: appwrite-jwt, data is the JWT string) or a session (authentication-method: appwrite-session, data is the session store blob, the same value a client holds as its session cookie).What each packet handles
min(publish QoS, granted QoS)and tracks each QoS 1 delivery until its PUBACK.Connection lifecycle
A connection is created on the first packet from a file descriptor. CONNECT authenticates it and marks it active. Once active, the broker schedules it in the keep-alive wheel.
The keep-alive wheel is a hashed timing wheel. Each connection sits in a bucket for its deadline second. A single timer runs every 20 seconds, drains the buckets that came due, and closes any client that has been silent for longer than
keepAlive * 1.5. Client activity only moves a timestamp on the connection. It does not touch the wheel, so a busy client does not churn it.Close is centralized. Whether the client sends DISCONNECT, the keep-alive reaper fires, or the socket drops, Swoole's onClose calls one method that unsubscribes the file descriptor, removes it from the wheel, and decrements the active gauge.
The whole packet lifecycle in onReceive is wrapped. A malformed packet or a handler failure is logged and closes only that connection. It never propagates and takes down the Swoole worker.
Offline replay: ledger, cursor, trie
Ledger. Every published message is appended to the
appwritePushLedgercollection with a per topic sequence number. The sequence comes from an atomic increment ontopics.sequence. The increment and the insert run in one transaction, opened by locking the topic row, so two concurrent publishes cannot open a gap and a retry is idempotent on(messageId, topic).Cursor. Each client has a cursor per topic in Redis, keyed by project, user, and client id. The cursor is the highest sequence the client has acknowledged. A PUBACK advances it. Nothing else does.
Replay. On SUBSCRIBE, the broker reads the ledger from
cursor + 1up to the current topic sequence, capped at a replay depth, and re-sends those messages with the DUP flag set. The depth comes from the user's org plan (getPlanForUser, self hosted default of 5 from_APP_MQTT_REPLAY_DEPTH). Replay is a QoS 1 guarantee, so a QoS 0 subscription is live only and is not replayed. A clean-start connection purges its cursor first.Retention. The maintenance worker deletes ledger rows older than 7 days, which matches the maximum replay window.
Trie.
SubscriptionStoreis a tree keyed by project and topic segments. It resolves the set of file descriptors subscribed to a topic, each with its granted QoS, which is what the fan-out path iterates.Security
_APP_OPTIONS_ABUSE. It is keyed on userId for now because the infra does not surface the client IP to the broker.Telemetry
New instruments cover auth duration, connection duration, message size, bytes in and out, reauth outcomes, and PUBACK counts, alongside the existing connection and delivery counters.
Tests and CI
Tests / E2E / MQTTCI job boots the broker and runs themqttgroup. The general service lanes exclude that group so the broker dependent tests only run where the broker is guaranteed up.Connectivity (broker topology)
Clients reach the broker two ways, both TLS-only and terminated at Traefik (the broker itself stays plaintext on the internal Docker network, mirroring how realtime's
wssis terminated at the edge). The transport is addressed by host + dedicated port — there is no URL path — so a single push endpoint (e.g.push.cloud.appwrite.io) serves both native MQTT and MQTT-over-WebSocket:mqtts://(native MQTT)appwrite_mqtt:8883HostSNI(*), tls:1883wss://(MQTT over WebSocket)appwrite_mqtt_ws:8084:8083mqtts://<host>:8883; browsers usewss://<host>:8084(MQTT 5.0 framed in WebSocket binary frames)._APP_MQTT_PORT(8883) and_APP_MQTT_WSS_PORT(8084); the broker's internal WebSocket listener is_APP_MQTT_WS_PORT(8083).