Skip to content

feat(Push): add Appwrite Push (MQTT 5) adapter - #129

Open
deepshekhardas wants to merge 1 commit into
utopia-php:mainfrom
deepshekhardas:fix/122-appwrite-push-mqtt5
Open

feat(Push): add Appwrite Push (MQTT 5) adapter#129
deepshekhardas wants to merge 1 commit into
utopia-php:mainfrom
deepshekhardas:fix/122-appwrite-push-mqtt5

Conversation

@deepshekhardas

Copy link
Copy Markdown

Port of PR #122 by abnegate.

Adds Appwrite Push - a self-hosted, low-power alternative to FCM/APNS that publishes notifications over MQTT 5 to per-device topics.

Changes:

  • New MQTT 5 control-packet codec (Helpers/MQTT) - pure PHP, no extra dependency
  • New Appwrite Push adapter for MQTT 5 publishing
  • Fake broker for integration testing
  • Unit and integration tests

@greptile-apps

greptile-apps Bot commented Jun 14, 2026

Copy link
Copy Markdown

Greptile Summary

This PR adds a new Appwrite Push adapter that publishes MQTT 5 notifications to a self-hosted broker, along with a pure-PHP MQTT 5 codec (Helpers/MQTT), a Swoole-based fake broker for integration testing, and accompanying unit and integration tests.

  • Adapter/Push/Appwrite.php: Connects over TLS, authenticates with a short-lived HMAC-signed JWT, and pipelines QoS-1 PUBLISHes across up to receiveMaximum in-flight packets for throughput. The pipelinedPublish loop's error path only records currently-in-flight tokens when the socket fails, silently omitting all tokens that haven't been sent yet from the response.
  • Helpers/MQTT.php: Implements the minimal MQTT 5 control-packet subset needed for publishing. The readProperties switch exits early on any unrecognised property ID, which can silently drop receiveMaximum from a real broker's CONNACK if the broker sends assignedClientIdentifier (a common MQTT 5 property) first.
  • tests/: Good round-trip coverage of the codec and adapter; the integration tests require the Swoole extension but have no skip guard, producing a cryptic timeout failure when Swoole is absent.

Confidence Score: 3/5

  • The adapter has a functional gap in its error-recovery path where a mid-batch socket failure causes an unknown number of device tokens to be silently dropped from the response, combined with a known codec issue where real-broker CONNACK properties can be partially parsed.
  • When the broker closes the connection (or times out) while pipelinedPublish is draining ACKs, only the currently in-flight window of tokens is marked as failed — any tokens that haven't been sent yet are never recorded in the response. A fan-out to 5,000 devices that loses its connection after the first 256 would return a response showing only 256 results with no indication that 4,744 were never attempted. Additionally, the MQTT codec's readProperties parser exits early on unknown property IDs, meaning a real broker that sends assignedClientIdentifier before receiveMaximum in its CONNACK (standard behavior for many brokers) would cause the adapter to silently operate with the wrong flow-control window.
  • src/Utopia/Messaging/Adapter/Push/Appwrite.php (pipelinedPublish error path) and src/Utopia/Messaging/Helpers/MQTT.php (readProperties switch) need attention before merging.

Important Files Changed

Filename Overview
src/Utopia/Messaging/Adapter/Push/Appwrite.php New MQTT 5 push adapter; pipelined publish loop silently drops unsent tokens from the response when a socket error occurs mid-batch. Several previously-flagged issues also remain (readBuffer not reset between calls, rtrim vs trim, receiveMaximum monotonically decreasing).
src/Utopia/Messaging/Helpers/MQTT.php Pure-PHP MQTT 5 codec; readProperties returns early on any unknown property ID, silently discarding all subsequent properties in the same packet — including receiveMaximum when preceded by assignedClientId in a real broker's CONNACK (previously flagged).
tests/Messaging/Adapter/Push/AppwriteTest.php Integration tests cover happy-path, pipelining, and partial-failure; relies on Swoole extension without a skip guard, causing opaque failures when Swoole is absent. State files written by startBroker are also not cleaned up.
tests/Messaging/Adapter/Push/FakeBroker.php Swoole-based in-process MQTT broker for integration tests; correctly handles CONNECT/PUBLISH/PINGREQ/DISCONNECT and flushes capture state on each event. Does not validate JWTs, which is intentional for test isolation.
tests/Messaging/Helpers/MQTTTest.php Good unit coverage of the MQTT codec: round-trip encode/decode for CONNECT, PUBLISH, CONNACK, PUBACK, PINGREQ/RESP; partial-buffer and multi-packet-coalescing edge cases; validation error paths.

Fix All in Claude Code Fix All in Codex

Prompt To Fix All With AI
### Issue 1
src/Utopia/Messaging/Adapter/Push/Appwrite.php:151-156
**Unsent tokens silently absent from response on socket failure**

When `readPacket` throws (broker timeout, EOF, or disconnect) only the currently-inflight tokens are recorded as failures. Any tokens with index `$cursor` through `$total - 1` that have not yet been sent are never added to `$response`, so they are invisible to the caller — they appear neither as successes nor failures. With a 5,000-token fan-out and a broker that closes the socket after the first window, ~4,744 tokens would silently vanish from the returned results array.

The fix is to also record the unsent remainder before returning:

```php
} catch (\Throwable $error) {
    foreach ($inflight as $token) {
        $response->addResult($token, $error->getMessage());
    }
    // Mark tokens that were never attempted.
    for ($i = $cursor; $i < $total; $i++) {
        $response->addResult($tokens[$i], $error->getMessage());
    }
    return;
}
```

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Reviews (6): Last reviewed commit: "feat(Push): add Appwrite Push (MQTT 5) a..." | Re-trigger Greptile

}

public function getMaxMessagesPerRequest(): int
{

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 readBuffer not cleared between process() calls

$this->readBuffer is never reset at the start of each connection. If the adapter instance is reused (e.g., send() is called twice), or if the broker sends an extra packet after the last PUBACK (e.g., a PINGREQ that landed in the buffer just before disconnect), that residual data persists into the next call. On the next invocation readPacket() would immediately return the leftover packet as if it were the new connection's CONNACK, causing handshake() to throw "Broker did not respond with CONNACK" even on a healthy connection.

Add $this->readBuffer = ''; at the start of connect() or at the top of process() to isolate each connection's read state.


private function resolveEndpoint(): string
{
$endpoint = \rtrim($this->endpoint);

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 rtrim strips only trailing whitespace, so a leading space in the configured endpoint (e.g., " broker.example.com") would produce a malformed URL like tls:// broker.example.com:8883 that stream_socket_client rejects. Use trim to strip both ends.

Suggested change
$endpoint = \rtrim($this->endpoint);
$endpoint = \trim($this->endpoint);

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!

Comment on lines +381 to +384
$packet = MQTT::decodePacket($this->readBuffer);
if ($packet !== null) {
return $packet;
}

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 receiveMaximum decreases monotonically across process() calls

$this->receiveMaximum is instance state that is only ever updated via min() in handshake(). If the adapter is reused across multiple send() calls and the broker advertises a low receiveMaximum (say 10) on the first call, subsequent connections — even to a different broker endpoint — will be throttled to that minimum permanently for the lifetime of the object. Resetting it to the class-default (or to 65535) at the start of each connect() would make each connection's window independent.

@deepshekhardas

Copy link
Copy Markdown
Author

Following up - this PR has been open for 1 month. Let me know if any changes are needed or if the implementation approach needs adjustment.

Based on PR utopia-php#122 by abnegate. Adds Appwrite Push - a self-hosted MQTT 5 based push notification adapter with minimal MQTT 5 control-packet codec.
Comment on lines +151 to +156
} catch (\Throwable $error) {
foreach ($inflight as $token) {
$response->addResult($token, $error->getMessage());
}
return;
}

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Unsent tokens silently absent from response on socket failure

When readPacket throws (broker timeout, EOF, or disconnect) only the currently-inflight tokens are recorded as failures. Any tokens with index $cursor through $total - 1 that have not yet been sent are never added to $response, so they are invisible to the caller — they appear neither as successes nor failures. With a 5,000-token fan-out and a broker that closes the socket after the first window, ~4,744 tokens would silently vanish from the returned results array.

The fix is to also record the unsent remainder before returning:

} catch (\Throwable $error) {
    foreach ($inflight as $token) {
        $response->addResult($token, $error->getMessage());
    }
    // Mark tokens that were never attempted.
    for ($i = $cursor; $i < $total; $i++) {
        $response->addResult($tokens[$i], $error->getMessage());
    }
    return;
}
Prompt To Fix With AI
This is a comment left during a code review.
Path: src/Utopia/Messaging/Adapter/Push/Appwrite.php
Line: 151-156

Comment:
**Unsent tokens silently absent from response on socket failure**

When `readPacket` throws (broker timeout, EOF, or disconnect) only the currently-inflight tokens are recorded as failures. Any tokens with index `$cursor` through `$total - 1` that have not yet been sent are never added to `$response`, so they are invisible to the caller — they appear neither as successes nor failures. With a 5,000-token fan-out and a broker that closes the socket after the first window, ~4,744 tokens would silently vanish from the returned results array.

The fix is to also record the unsent remainder before returning:

```php
} catch (\Throwable $error) {
    foreach ($inflight as $token) {
        $response->addResult($token, $error->getMessage());
    }
    // Mark tokens that were never attempted.
    for ($i = $cursor; $i < $total; $i++) {
        $response->addResult($tokens[$i], $error->getMessage());
    }
    return;
}
```

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Fix in Claude Code Fix in Codex

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant