feat(messaging): add Appwrite Push adapter for MQTT 5 integration by ArnabChatterjee20k · Pull Request #161 · utopia-php/monorepo

ArnabChatterjee20k · 2026-08-25T10:55:33Z

Appwrite Push (MQTT 5) adapter

Adds a Push adapter that talks to Appwrite's custom MQTT 5 broker, a minimal MQTT 5 control-packet codec (Helpers/MQTT), and a spawned mock broker for tests.

The adapter speaks the broker's enhanced-authentication CONNECT dialect: the credential rides in the Authentication Method/Data properties and the project in a projectId User Property. The broker scopes every topic by that project, so callers use bare device topics.

Publish — `send()`

use Utopia\Messaging\Adapter\Push\Appwrite;
use Utopia\Messaging\Messages\Push;

$adapter = new Appwrite(
    endpoint: 'broker.example.com:1883',
    projectId: '<projectId>',
    credential: $jwt,            // an Appwrite JWT (or a session secret)
    authMethod: 'appwrite-jwt',  // or 'appwrite-session'
    tls: true,
);

$response = $adapter->send(new Push(
    to: ['device-token-1', 'device-token-2'], // each published to appwrite/push/{token}
    title: 'Hello',
    body: 'World',
    data: ['key' => 'value'],
));

// $response['deliveredTo'], $response['results'][n]['status'] ...

Publishes are pipelined at QoS 1 (send a window of PUBLISHes, drain PUBACKs by packet id, refill), so throughput scales with socket bandwidth rather than round-trip latency.

Consume — `consume()`

The same enhanced-auth connection can subscribe. consume() subscribes to the given topics and invokes a callback for each message, until a message limit or timeout is reached (QoS 1 messages are acked):

$adapter->consume(
    topics: ['appwrite/push/device-token-1'],
    onMessage: function (array $message): void {
        // $message['topic'], $message['payload'], $message['qos']
    },
    limit: 1,
    timeout: 5.0,
);

This is what verifies broker fan-out end to end from the consumer side (e.g. a device receiving what a publisher sent).

Telemetry

An optional Telemetry adapter can be injected; it defaults to the base no-op counter:

new Appwrite(/* ... */, telemetry: $telemetry);

Codec — `Helpers/MQTT`

A transport-agnostic MQTT 5 codec: encode/parse CONNECT, CONNACK, PUBLISH, PUBACK, SUBSCRIBE, SUBACK, PINGREQ/RESP, DISCONNECT, and the v5 property block (User Properties, enhanced-auth, message expiry, …). Both the adapter and the tests build on it.

Tests

tests/Messaging/Adapter/Push/AppwriteTest.php drives the adapter against a spawned in-process mock broker (FakeBroker.php) that accepts the enhanced-auth CONNECT and acks QoS 1 publishes: device-topic publishes, pipelined PUBACK counting, expired-token reason codes, and the CONNECT property block (projectId + credential, not username/password).

Align the ported MQTT 5 codec and Appwrite Push adapter with the monorepo's stricter toolchain: pint (per preset, native_function_invocation, cast spacing, trailing commas), rector 2.x (explicit bool compare, drop redundant casts, empty()->=== [], readonly promoted props, instanceof over !== null), and phpstan 2.x (cast unpack() result feeding the by-ref int $offset). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Add a spawned in-process MQTT 5 broker (FakeBroker) that speaks just enough protocol to accept the adapter's enhanced-auth CONNECT, ack QoS 1 PUBLISHes (optionally rejecting tokens), and record what it saw. AppwriteTest drives the publisher against it: device-topic publishes, pipelined PUBACK counting, expired-token reason codes, and the enhanced-auth CONNECT (projectId + credential in the property block, not username/password). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

encodeSubscribe() completes the subscriber role of the codec: one topic filter, a configurable max QoS, and optional User Properties (e.g. the subId the broker keys subscriptions on). Reserved fixed-header flags set to 0b0010 per spec. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

…pter The same enhanced-auth connection that publishes can now subscribe: consume() connects, subscribes to the given topics (subId as a User Property), and invokes a callback for each PUBLISH until a message limit or timeout is reached, acking QoS 1. This is what verifies broker fan-out end to end from the consumer side. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

The adapter overrode the constructor without chaining to the base Adapter, so the send counter was never initialized and send() failed with "Typed property $sendCounter must not be accessed before initialization". Call parent::__construct and accept an optional Telemetry to inject, defaulting to the base no-op counter. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Add MQTT::parseSuback() and have consume() inspect the SUBACK reason codes: a code >= 0x80 (e.g. 0x87 Not Authorized from a broker ACL) now throws instead of silently subscribing to nothing, so subscribe-side authorization is observable. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

…sume()

greptile-apps · 2026-09-09T09:05:27Z

Confidence Score: 2/5

The PR is not safe to merge until per-recipient failure reporting, MQTT property negotiation, cross-connection decoder state, and the explicit testing-rule violation are addressed.

Findings

Unsent Recipients Lose Results ▶
Unknown Properties Stop Parsing ▶
Socket State Leaks Across Connections ▶
Tests Echo Production Details ▶

Summary

Publishes per-device JSON notifications through pipelined QoS 1 MQTT messages.
Adds enhanced-authentication CONNECT and subscription/consumption support.
Introduces MQTT packet and property encoding/parsing helpers.
Adds publisher tests backed by a Swoole-based fake broker.
The review identified incomplete failure accounting, fragile CONNACK property handling, cross-connection decoder state, and tests coupled to the production implementation.

greptile-apps · 2026-09-09T09:05:30Z

+            try {
+                $ack = $this->readPacket($socket);
+            } catch (\Throwable $error) {
+                foreach ($inflight as $token) {
+                    $response->addResult($token, $error->getMessage());
+                }
+                return;
+            }


Unsent Recipients Lose Results

If a socket read fails before all recipients enter the in-flight window, this catch records failures only for the current window and then returns. A request can contain 5,000 recipients while the window is at most 256, so later recipients silently disappear from results. Callers therefore cannot identify which notifications were never sent.

Knowledge Base Used: Messaging, queues, and NATS

Prompt To Fix With AI

This is a comment left during a code review. Path: packages/messaging/src/Utopia/Messaging/Adapter/Push/Appwrite.php Line: 239-246 Comment: **Unsent Recipients Lose Results** If a socket read fails before all recipients enter the in-flight window, this catch records failures only for the current window and then returns. A request can contain 5,000 recipients while the window is at most 256, so later recipients silently disappear from `results`. Callers therefore cannot identify which notifications were never sent. **Knowledge Base Used:** [Messaging, queues, and NATS](https://app.greptile.com/appwrite/-/custom-context/knowledge-base/utopia-php/monorepo/-/docs/messaging-queue-nats.md) --- For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

greptile-apps · 2026-09-09T09:05:30Z

+                    break;
+                default:
+                    return $properties;


Unknown Properties Stop Parsing

A valid but unimplemented MQTT property makes this parser abandon the rest of the property block. Because CONNACK properties may appear in any order, a property such as Maximum QoS or Retain Available can appear before Receive Maximum. The adapter then misses a broker limit below 256, sends too many unacknowledged QoS 1 messages, and can be disconnected for exceeding the negotiated window.

Knowledge Base Used: Asynchronous workflows and delivery

Prompt To Fix With AI

This is a comment left during a code review. Path: packages/messaging/src/Utopia/Messaging/Helpers/MQTT.php Line: 665-667 Comment: **Unknown Properties Stop Parsing** A valid but unimplemented MQTT property makes this parser abandon the rest of the property block. Because CONNACK properties may appear in any order, a property such as Maximum QoS or Retain Available can appear before Receive Maximum. The adapter then misses a broker limit below 256, sends too many unacknowledged QoS 1 messages, and can be disconnected for exceeding the negotiated window. **Knowledge Base Used:** [Asynchronous workflows and delivery](https://app.greptile.com/appwrite/-/custom-context/knowledge-base/utopia-php/monorepo/-/docs/asynchronous-workflows.md) --- For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

greptile-apps · 2026-09-09T09:05:30Z

+    private function connect()
+    {
+        $url = $this->resolveEndpoint();
+        $context = stream_context_create([
+            'ssl' => [
+                'verify_peer' => true,
+                'verify_peer_name' => true,
+                'SNI_enabled' => true,
+            ],
+        ]);
+
+        $socket = @stream_socket_client(
+            $url,
+            $errno,
+            $errstr,
+            self::CONNECT_TIMEOUT,
+            STREAM_CLIENT_CONNECT,
+            $context,
+        );
+
+        if (!$socket) {
+            throw new \RuntimeException("Unable to connect to Appwrite Push broker at {$url}: {$errstr} (errno {$errno})");
+        }
+
+        stream_set_timeout($socket, self::READ_TIMEOUT);
+
+        return $socket;
+    }


Socket State Leaks Across Connections

Connection-specific decoder state remains on the adapter when a new socket is opened. If consume() reaches its limit while another complete or partial PUBLISH remains in readBuffer, the next send() or consume() reads those old bytes during its handshake. It can then mistake a stale PUBLISH for CONNACK or combine bytes from different streams, causing the new connection to fail.

Knowledge Base Used: Asynchronous workflows and delivery

Prompt To Fix With AI

This is a comment left during a code review. Path: packages/messaging/src/Utopia/Messaging/Adapter/Push/Appwrite.php Line: 360-387 Comment: **Socket State Leaks Across Connections** Connection-specific decoder state remains on the adapter when a new socket is opened. If `consume()` reaches its limit while another complete or partial PUBLISH remains in `readBuffer`, the next `send()` or `consume()` reads those old bytes during its handshake. It can then mistake a stale PUBLISH for CONNACK or combine bytes from different streams, causing the new connection to fail. **Knowledge Base Used:** [Asynchronous workflows and delivery](https://app.greptile.com/appwrite/-/custom-context/knowledge-base/utopia-php/monorepo/-/docs/asynchronous-workflows.md) --- For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

greptile-apps · 2026-09-09T09:05:30Z

+            $this->assertSame('appwrite/push/device-token-1', $captured['publishes'][0]['topic']);
+            $this->assertSame('appwrite/push/device-token-2', $captured['publishes'][1]['topic']);
+
+            $decoded = json_decode($captured['publishes'][0]['payload'], true);
+            $this->assertSame('Hi', $decoded['notification']['title']);
+            $this->assertSame('Hello', $decoded['notification']['body']);
+            $this->assertSame(['k' => 'v'], $decoded['data']);
+
+            // Enhanced-auth CONNECT: credential + project ride the property block, not username/password.
+            $this->assertSame(self::PROJECT, $captured['connect']['projectId']);
+            $this->assertSame('appwrite-jwt', $captured['connect']['authMethod']);
+            $this->assertSame(self::CREDENTIAL, $captured['connect']['credential']);
+            $this->assertStringStartsWith('appwrite-server-', $captured['connect']['clientId']);


Tests Echo Production Details

These assertions copy source-defined values such as appwrite/push and appwrite-server-, while the fake broker decodes and acknowledges traffic with the same production MQTT codec as the adapter. Both sides can therefore share an invalid wire encoding while the tests remain green. This violates the repository directive to test observable behavior rather than mirror source code or configuration, so the requirement must be satisfied before merging. Use an independent protocol boundary or observable end-to-end delivery test instead; the same coupling also appears in the expected-topic construction at lines 102–104.

Context Used: Call out and harshly judge implementation-coupled ... (source)

Prompt To Fix With AI

This is a comment left during a code review. Path: packages/messaging/tests/Messaging/Adapter/Push/AppwriteTest.php Line: 50-62 Comment: **Tests Echo Production Details** These assertions copy source-defined values such as `appwrite/push` and `appwrite-server-`, while the fake broker decodes and acknowledges traffic with the same production MQTT codec as the adapter. Both sides can therefore share an invalid wire encoding while the tests remain green. This violates the repository directive to test observable behavior rather than mirror source code or configuration, so the requirement must be satisfied before merging. Use an independent protocol boundary or observable end-to-end delivery test instead; the same coupling also appears in the expected-topic construction at lines 102–104. **Context Used:** Call out and harshly judge implementation-coupled ... ([source](https://app.greptile.com/review/custom-context?memory=instruction-0)) --- For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

feat(messaging): add Appwrite Push adapter for MQTT 5 integration

4f821cc

ArnabChatterjee20k requested review from Meldiron, abnegate, eldadfux and loks0n as code owners August 25, 2026 10:55

ArnabChatterjee20k and others added 7 commits August 25, 2026 16:37

fix(messaging): remove userProperties from subscription packet in con…

2620971

…sume()

greptile-apps Bot reviewed Sep 9, 2026

View reviewed changes

Navigation Menu

Uh oh!

feat(messaging): add Appwrite Push adapter for MQTT 5 integration - #161

feat(messaging): add Appwrite Push adapter for MQTT 5 integration#161
ArnabChatterjee20k wants to merge 8 commits into
mainfrom
appwrite-mqtt-poc

ArnabChatterjee20k commented Aug 25, 2026 •

edited

Loading

Uh oh!

greptile-apps Bot commented Sep 9, 2026 •

edited

Loading

Summary

Uh oh!

greptile-apps Bot Sep 9, 2026

Uh oh!

greptile-apps Bot Sep 9, 2026

Uh oh!

greptile-apps Bot Sep 9, 2026

Uh oh!

greptile-apps Bot Sep 9, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Uh oh!

Conversation

ArnabChatterjee20k commented Aug 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Appwrite Push (MQTT 5) adapter

Publish — send()

Consume — consume()

Telemetry

Codec — Helpers/MQTT

Tests

Uh oh!

greptile-apps Bot commented Sep 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Confidence Score: 2/5

Findings

Summary

Uh oh!

greptile-apps Bot Sep 9, 2026

Choose a reason for hiding this comment

Uh oh!

greptile-apps Bot Sep 9, 2026

Choose a reason for hiding this comment

Uh oh!

greptile-apps Bot Sep 9, 2026

Choose a reason for hiding this comment

Uh oh!

greptile-apps Bot Sep 9, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

ArnabChatterjee20k commented Aug 25, 2026 •

edited

Loading

Publish — `send()`

Consume — `consume()`

Codec — `Helpers/MQTT`

greptile-apps Bot commented Sep 9, 2026 •

edited

Loading