Merge pull request #11767 from appwrite/realtime-query-message-payload

Realtime query message payload

app/realtime.php

@@ -450,7 +450,7 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats,
                         ]
                     ];
 
-                    $server->send($realtime->getSubscribers($event), json_encode([
+                    $server->send(array_keys($realtime->getSubscribers($event)), json_encode([
                         'type' => 'event',
                         'data' => $event['data']
                     ]));
@@ -712,11 +712,43 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
 
         $channels = Realtime::convertChannels($request->getQuery('channels', []), $user->getId());
 
+        $updateStats = static function (string $projectId, ?string $teamId, string $payloadJson) use ($register, $stats): void {
+            $register->get('telemetry.connectionCounter')->add(1);
+            $register->get('telemetry.connectionCreatedCounter')->add(1);
+
+            $stats->set($projectId, [
+                'projectId' => $projectId,
+                'teamId' => $teamId
+            ]);
+            $stats->incr($projectId, 'connections');
+            $stats->incr($projectId, 'connectionsTotal');
+
+            triggerStats([
+                METRIC_REALTIME_CONNECTIONS => 1,
+                METRIC_REALTIME_OUTBOUND => \strlen($payloadJson),
+            ], $projectId);
+        };
+
         /**
          * Channels Check
          */
         if (empty($channels)) {
-            throw new Exception(Exception::REALTIME_POLICY_VIOLATION, 'Missing channels');
+            // in case of message based 'subscribe' channels will be empty at first and only projectId and roles will be available
+            $sanitizedUser = empty($user->getId()) ? null : $response->output($user, Response::MODEL_ACCOUNT);
+            $connectedPayloadJson = json_encode([
+                'type' => 'connected',
+                'data' => [
+                    'channels' => [],
+                    'subscriptions' => [],
+                    'user' => $sanitizedUser
+                ]
+            ]);
+
+            $realtime->subscribe($project->getId(), $connection, '', $roles, [], [], $user->getId());
+            $realtime->connections[$connection]['authorization'] = $authorization;
+            $server->send([$connection], $connectedPayloadJson);
+            $updateStats($project->getId(), $project->getAttribute('teamId'), $connectedPayloadJson);
+            return;
         }
 
         $names = array_keys($channels);
@@ -740,7 +772,8 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
                 $subscriptionId,
                 $roles,
                 $subscription['channels'],
-                $subscription['queries']
+                $subscription['queries'],
+                $user->getId()
             );
 
             $mapping[$index] = $subscriptionId;
@@ -760,20 +793,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
         ]);
 
         $server->send([$connection], $connectedPayloadJson);
-
-        $register->get('telemetry.connectionCounter')->add(1);
-        $register->get('telemetry.connectionCreatedCounter')->add(1);
-
-        $stats->set($project->getId(), [
-            'projectId' => $project->getId(),
-            'teamId' => $project->getAttribute('teamId')
-        ]);
-        $stats->incr($project->getId(), 'connections');
-        $stats->incr($project->getId(), 'connectionsTotal');
-
-        $connectedOutboundBytes = \strlen($connectedPayloadJson);
-
-        triggerStats([METRIC_REALTIME_CONNECTIONS => 1, METRIC_REALTIME_OUTBOUND => $connectedOutboundBytes], $project->getId());
+        $updateStats($project->getId(), $project->getAttribute('teamId'), $connectedPayloadJson);
 
 
     } catch (Throwable $th) {
@@ -815,7 +835,6 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
 $server->onMessage(function (int $connection, string $message) use ($server, $realtime, $containerId) {
     $project = null;
     $authorization = null;
-
     try {
         $rawSize = \strlen($message);
         $response = new Response(new SwooleResponse());
@@ -941,7 +960,8 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
                             $subscriptionId,
                             $roles,
                             $subscription['channels'] ?? [],
-                            $queries
+                            $queries,
+                            $user->getId()
                         );
                     }
                 }
@@ -975,6 +995,99 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
 
                 break;
 
+            case 'subscribe':
+                /**
+                 * Message based upsertion of a subscription
+                 * If subscriptionId is given then it will match subId of the connection and update the subscription with channels and queries
+                 * If non-existing subid is given or not given a new subid will be generated
+                 * Similar to what we have now -> two subscribe() block with same channels and queries still two different subscriptions
+                 *
+                 * structure of the payload -> array of maps
+                 * 'data' : [subscriptionId:"" , channels:[] , queries:[]]
+                 */
+                if (!is_array($message['data']) || !array_is_list($message['data'])) {
+                    throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Payload is not valid.');
+                }
+
+                $roles = $realtime->connections[$connection]['roles'] ?? [Role::guests()->toString()];
+                $userId = $realtime->connections[$connection]['userId'] ?? '';
+
+                // bulk validation + parsing before subscribing
+                $parsedPayloads = [];
+                foreach ($message['data'] as $payload) {
+                    if (!\is_array($payload)) {
+                        throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Each subscribe payload must be an object.');
+                    }
+                    if (!array_key_exists('channels', $payload)) {
+                        throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'channels is not present in payload.');
+                    }
+                    if (!is_array($payload['channels']) || !array_is_list($payload['channels'])) {
+                        throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'channels is not a valid array.');
+                    }
+                    // registering the queries if not present and check in the same payload later on
+                    if (!array_key_exists('queries', $payload)) {
+                        $payload['queries'] = [];
+                    }
+                    if (!is_array($payload['queries']) || !array_is_list($payload['queries'])) {
+                        throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'queries is not a valid array.');
+                    }
+
+                    $subscriptionId = \array_key_exists('subscriptionId', $payload)
+                        ? $payload['subscriptionId']
+                        : ID::unique();
+
+                    try {
+                        $convertedQueries = Realtime::convertQueries($payload['queries']);
+                    } catch (QueryException $e) {
+                        throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Invalid query: ' . $e->getMessage());
+                    }
+
+                    $parsedPayloads[] = [
+                        'subscriptionId' => $subscriptionId,
+                        'channels' => $payload['channels'],
+                        'queries' => $convertedQueries,
+                    ];
+                }
+
+                foreach ($parsedPayloads as $parsedPayload) {
+                    $subscriptionId = $parsedPayload['subscriptionId'];
+                    $channels = \array_keys(Realtime::convertChannels($parsedPayload['channels'], $userId));
+                    $queries = $parsedPayload['queries'];
+                    $realtime->subscribe($projectId, $connection, $subscriptionId, $roles, $channels, $queries);
+                }
+
+                // subscribe() overwrites the connection entry; restore auth so later onMessage uses the same context.
+                $realtime->connections[$connection]['authorization'] = $authorization;
+
+                $responsePayload = json_encode([
+                    'type' => 'response',
+                    'data' => [
+                        'to' => 'subscribe',
+                        'success' => true,
+                        'subscriptions' => \array_map(function (array $parsedPayload) {
+                            return [
+                                'subscriptionId' => $parsedPayload['subscriptionId'],
+                                'channels' => $parsedPayload['channels'],
+                                'queries' => \array_map(fn ($q) => $q->toString(), $parsedPayload['queries']),
+                            ];
+                        }, $parsedPayloads),
+                    ]
+                ]);
+
+                $server->send([$connection], $responsePayload);
+
+                if ($project !== null && !$project->isEmpty()) {
+                    $subscribeOutboundBytes = \strlen($responsePayload);
+
+                    if ($subscribeOutboundBytes > 0) {
+                        triggerStats([
+                            METRIC_REALTIME_OUTBOUND => $subscribeOutboundBytes,
+                        ], $project->getId());
+                    }
+                }
+
+                break;
+
             default:
                 throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Message type is not valid.');
         }

src/Appwrite/Messaging/Adapter/Realtime.php

@@ -20,6 +20,7 @@ class Realtime extends MessagingAdapter
      * [CONNECTION_ID] ->
      *      'projectId' -> [PROJECT_ID]
      *      'roles' -> [ROLE_x, ROLE_Y]
+     *      'userId' -> [USER_ID]
      *      'channels' -> [CHANNEL_NAME_X, CHANNEL_NAME_Y, CHANNEL_NAME_Z]
      */
     public array $connections = [];
@@ -67,25 +68,35 @@ class Realtime extends MessagingAdapter
      * @param array $queryGroup Array of Query objects for this subscription (AND logic within subscription)
      * @return void
      */
-    public function subscribe(string $projectId, mixed $identifier, string $subscriptionId, array $roles, array $channels, array $queryGroup = []): void
-    {
+    public function subscribe(
+        string $projectId,
+        mixed $identifier,
+        string $subscriptionId,
+        array $roles,
+        array $channels,
+        array $queryGroup = [],
+        ?string $userId = null
+    ): void {
         if (!isset($this->subscriptions[$projectId])) { // Init Project
             $this->subscriptions[$projectId] = [];
         }
 
         $strings = [];
-        if (empty($queryGroup)) {
-            $strings[] = Query::select(['*'])->toString();
-        } else {
-            foreach ($queryGroup as $query) {
-                $strings[] = $query->toString();
-            }
-        }
+        $data = [];
 
-        $data = [
-            'strings' => $strings,
-            'compiled' => RuntimeQuery::compile($queryGroup),
-        ];
+        if (!empty($channels)) {
+            if (empty($queryGroup)) {
+                $strings[] = Query::select(['*'])->toString();
+            } else {
+                foreach ($queryGroup as $query) {
+                    $strings[] = $query->toString();
+                }
+            }
+            $data = [
+                'strings' => $strings,
+                'compiled' => RuntimeQuery::compile($queryGroup),
+            ];
+        }
 
         foreach ($roles as $role) {
             if (!isset($this->subscriptions[$projectId][$role])) {
@@ -103,10 +114,12 @@ class Realtime extends MessagingAdapter
             }
         }
 
-        // Update connection info
+        // Keep userId from onOpen/authentication when provided.
+        // Fallback to existing stored value for subsequent subscribe upserts.
         $this->connections[$identifier] = [
             'projectId' => $projectId,
             'roles' => $roles,
+            'userId' => $userId ?? ($this->connections[$identifier]['userId'] ?? ''),
             'channels' => $channels
         ];
     }

tests/e2e/Services/Realtime/RealtimeBase.php

@@ -101,18 +101,14 @@ trait RealtimeBase
         $client->close();
     }
 
-    public function testConnectionFailureMissingChannels(): void
+    public function testConnectionSuccessMissingChannels(): void
     {
         $client = $this->getWebsocket([]);
         $payload = json_decode($client->receive(), true);
 
         $this->assertArrayHasKey("type", $payload);
         $this->assertArrayHasKey("data", $payload);
-        $this->assertEquals("error", $payload["type"]);
-        $this->assertEquals(1008, $payload["data"]["code"]);
-        $this->assertEquals("Missing channels", $payload["data"]["message"]);
-        \usleep(250000); // 250ms
-        $this->expectException(ConnectionException::class); // Check if server disconnected client
+        $this->assertEquals("connected", $payload["type"]);
         $client->close();
     }
 

tests/e2e/Services/Realtime/RealtimeCustomClientQueryTestWithMessage.php

@@ -0,0 +1,521 @@
+<?php
+
+namespace Tests\E2E\Services\Realtime;
+
+use Tests\E2E\Client;
+use Tests\E2E\Scopes\ProjectCustom;
+use Tests\E2E\Scopes\Scope;
+use Tests\E2E\Scopes\SideClient;
+use Utopia\Database\Helpers\ID;
+use Utopia\Database\Helpers\Permission;
+use Utopia\Database\Helpers\Role;
+use Utopia\Database\Query;
+use WebSocket\Client as WebSocketClient;
+use WebSocket\TimeoutException;
+
+class RealtimeCustomClientQueryTestWithMessage extends Scope
+{
+    use ProjectCustom;
+    use SideClient;
+    use RealtimeQueryBase;
+
+    protected function supportForCheckConnectionStatus(): bool
+    {
+        return false;
+    }
+
+    /**
+     * Same signature as `RealtimeBase::getWebsocket()`, but:
+     * - never sends queries in the URL (avoids URL length limits)
+     * - once connected, sends channel/query data using a `type: "subscribe"` message
+     */
+    private function getWebsocket(
+        array $channels = [],
+        array $headers = [],
+        ?string $projectId = null,
+        ?array $queries = null,
+        int $timeout = 2
+    ): WebSocketClient {
+        if ($projectId === null) {
+            $projectId = $this->getProject()['$id'];
+        }
+
+        $queryString = \http_build_query([
+            'project' => $projectId,
+        ]);
+
+        $client = new WebSocketClient(
+            'ws://appwrite.test/v1/realtime?' . $queryString,
+            [
+                'headers' => $headers,
+                'timeout' => $timeout,
+            ]
+        );
+        $connected = \json_decode($client->receive(), true);
+        $this->assertEquals('connected', $connected['type'] ?? null);
+
+        if (empty($channels)) {
+            return $client;
+        }
+
+        if ($queries === []) {
+            $queries = [Query::select(['*'])->toString()];
+        }
+
+        $payload = [[
+            'channels' => $channels,
+        ]];
+
+        if ($queries !== null) {
+            $payload[0]['queries'] = $queries;
+        }
+
+        $existingSubscriptions = $connected['data']['subscriptions'] ?? [];
+        if (!empty($existingSubscriptions)) {
+            $payload[0]['subscriptionId'] = $existingSubscriptions[\array_key_first($existingSubscriptions)];
+        }
+
+        $client->send(\json_encode([
+            'type' => 'subscribe',
+            'data' => $payload,
+        ]));
+
+        $response = \json_decode($client->receive(), true);
+        $this->assertEquals('response', $response['type'] ?? null);
+        $this->assertEquals('subscribe', $response['data']['to'] ?? null);
+        $this->assertTrue($response['data']['success'] ?? false);
+        $this->assertArrayHasKey('subscriptions', $response['data']);
+        $this->assertIsArray($response['data']['subscriptions']);
+
+        return $client;
+    }
+
+    /**
+     * Connects (URL has no per-channel queries), then sends a subscribe message with the given query strings.
+     * Used to assert server rejects unsupported query methods the same way as URL-based subscriptions.
+     *
+     * @param  array<int, string>  $queryStrings
+     * @return array<string, mixed>
+     */
+    private function receiveSubscribeMessageResponse(
+        array $channels,
+        array $headers,
+        array $queryStrings
+    ): array {
+        $projectId = $this->getProject()['$id'];
+        $queryString = \http_build_query([
+            'project' => $projectId,
+        ]);
+
+        $client = new WebSocketClient(
+            'ws://appwrite.test/v1/realtime?' . $queryString,
+            [
+                'headers' => $headers,
+                'timeout' => 2,
+            ]
+        );
+        $connected = \json_decode($client->receive(), true);
+        $this->assertEquals('connected', $connected['type'] ?? null);
+
+        $client->send(\json_encode([
+            'type' => 'subscribe',
+            'data' => [[
+                'channels' => $channels,
+                'queries' => $queryStrings,
+            ]],
+        ]));
+
+        $response = \json_decode($client->receive(), true);
+        $client->close();
+
+        return $response;
+    }
+
+    private function getWebsocketWithCustomQuery(array $queryParams, array $headers = [], int $timeout = 2): WebSocketClient
+    {
+        $queryString = \http_build_query($queryParams);
+
+        return new WebSocketClient(
+            'ws://appwrite.test/v1/realtime?' . $queryString,
+            [
+                'headers' => $headers,
+                'timeout' => $timeout,
+            ]
+        );
+    }
+
+    /**
+     * @param  array<int, array<string, mixed>>  $payloadEntries
+     * @return array<string, mixed>
+     */
+    private function sendSubscribeMessage(WebSocketClient $client, array $payloadEntries): array
+    {
+        $client->send(\json_encode([
+            'type' => 'subscribe',
+            'data' => $payloadEntries,
+        ]));
+        $response = \json_decode($client->receive(), true);
+        $this->assertEquals('response', $response['type'] ?? null);
+        $this->assertEquals('subscribe', $response['data']['to'] ?? null);
+        $this->assertTrue($response['data']['success'] ?? false);
+        $this->assertArrayHasKey('subscriptions', $response['data']);
+        $this->assertIsArray($response['data']['subscriptions']);
+
+        return $response;
+    }
+
+    /**
+     * subscriptionId: update with id from connected, create by omitting id, explicit new id,
+     * duplicate id in one bulk (last wins), mixed bulk, idempotent repeat, empty queries → select-all.
+     */
+    public function testSubscribeMessageUpsertCreateAndEdgeCases(): void
+    {
+        $user = $this->getUser();
+        $session = $user['session'] ?? '';
+        $projectId = $this->getProject()['$id'];
+        $headers = [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $session,
+        ];
+
+        $queryString = \http_build_query([
+            'project' => $projectId,
+        ]);
+        $client = new WebSocketClient(
+            'ws://appwrite.test/v1/realtime?' . $queryString,
+            [
+                'headers' => $headers,
+                'timeout' => 30,
+            ]
+        );
+        $connected = \json_decode($client->receive(), true);
+        $this->assertEquals('connected', $connected['type'] ?? null);
+        $initialResponse = $this->sendSubscribeMessage($client, [[
+            'channels' => ['documents'],
+            'queries' => [Query::select(['*'])->toString()],
+        ]]);
+        $initialSubscriptionId = $initialResponse['data']['subscriptions'][0]['subscriptionId'] ?? '';
+        $this->assertNotEmpty($initialSubscriptionId);
+
+        $q1 = [Query::equal('status', ['q1'])->toString()];
+        $r1 = $this->sendSubscribeMessage($client, [[
+            'subscriptionId' => $initialSubscriptionId,
+            'channels' => ['documents'],
+            'queries' => $q1,
+        ]]);
+        $this->assertCount(1, $r1['data']['subscriptions']);
+        $this->assertSame($initialSubscriptionId, $r1['data']['subscriptions'][0]['subscriptionId']);
+        $this->assertSame($q1, $r1['data']['subscriptions'][0]['queries']);
+
+        $q2 = [Query::equal('status', ['q2'])->toString()];
+        $r2 = $this->sendSubscribeMessage($client, [[
+            'subscriptionId' => $initialSubscriptionId,
+            'channels' => ['documents'],
+            'queries' => $q2,
+        ]]);
+        $this->assertSame($initialSubscriptionId, $r2['data']['subscriptions'][0]['subscriptionId']);
+        $this->assertSame($q2, $r2['data']['subscriptions'][0]['queries']);
+
+        $rOmit = $this->sendSubscribeMessage($client, [[
+            'channels' => ['documents'],
+            'queries' => [Query::equal('status', ['omitted-slot'])->toString()],
+        ]]);
+        $mintedId = $rOmit['data']['subscriptions'][0]['subscriptionId'];
+        $this->assertNotSame($initialSubscriptionId, $mintedId);
+        $this->assertNotEmpty($mintedId);
+
+        $explicitNewId = ID::unique();
+        $qExplicit = [Query::equal('status', ['explicit'])->toString()];
+        $rExplicit = $this->sendSubscribeMessage($client, [[
+            'subscriptionId' => $explicitNewId,
+            'channels' => ['documents'],
+            'queries' => $qExplicit,
+        ]]);
+        $this->assertSame($explicitNewId, $rExplicit['data']['subscriptions'][0]['subscriptionId']);
+        $this->assertSame($qExplicit, $rExplicit['data']['subscriptions'][0]['queries']);
+
+        $qFirst = [Query::equal('status', ['dup-a'])->toString()];
+        $qSecond = [Query::equal('status', ['dup-b'])->toString()];
+        $rDup = $this->sendSubscribeMessage($client, [
+            [
+                'subscriptionId' => $initialSubscriptionId,
+                'channels' => ['documents'],
+                'queries' => $qFirst,
+            ],
+            [
+                'subscriptionId' => $initialSubscriptionId,
+                'channels' => ['documents'],
+                'queries' => $qSecond,
+            ],
+        ]);
+        $this->assertCount(2, $rDup['data']['subscriptions']);
+        $this->assertSame($initialSubscriptionId, $rDup['data']['subscriptions'][0]['subscriptionId']);
+        $this->assertSame($initialSubscriptionId, $rDup['data']['subscriptions'][1]['subscriptionId']);
+        $this->assertSame($qSecond, $rDup['data']['subscriptions'][1]['queries']);
+
+        $rMixed = $this->sendSubscribeMessage($client, [
+            [
+                'subscriptionId' => $initialSubscriptionId,
+                'channels' => ['documents'],
+                'queries' => [Query::equal('status', ['mixed-update'])->toString()],
+            ],
+            [
+                'channels' => ['documents'],
+                'queries' => [Query::equal('status', ['mixed-new'])->toString()],
+            ],
+        ]);
+        $this->assertCount(2, $rMixed['data']['subscriptions']);
+        $this->assertSame($initialSubscriptionId, $rMixed['data']['subscriptions'][0]['subscriptionId']);
+        $mixedSecondId = $rMixed['data']['subscriptions'][1]['subscriptionId'];
+        $this->assertNotSame($initialSubscriptionId, $mixedSecondId);
+        $this->assertNotEmpty($mixedSecondId);
+
+        $rSame = $this->sendSubscribeMessage($client, [[
+            'subscriptionId' => $initialSubscriptionId,
+            'channels' => ['documents'],
+            'queries' => [Query::equal('status', ['idempotent'])->toString()],
+        ]]);
+        $rSameAgain = $this->sendSubscribeMessage($client, [[
+            'subscriptionId' => $initialSubscriptionId,
+            'channels' => ['documents'],
+            'queries' => [Query::equal('status', ['idempotent'])->toString()],
+        ]]);
+        $this->assertSame($rSame['data']['subscriptions'][0]['queries'], $rSameAgain['data']['subscriptions'][0]['queries']);
+
+        $rEmpty = $this->sendSubscribeMessage($client, [[
+            'subscriptionId' => $initialSubscriptionId,
+            'channels' => ['documents'],
+            'queries' => [],
+        ]]);
+        $this->assertCount(1, $rEmpty['data']['subscriptions']);
+        $this->assertSame($initialSubscriptionId, $rEmpty['data']['subscriptions'][0]['subscriptionId']);
+
+        $client->close();
+    }
+
+    public function testInvalidQueryShouldNotSubscribe(): void
+    {
+        $user = $this->getUser();
+        $session = $user['session'] ?? '';
+        $projectId = $this->getProject()['$id'];
+        $headers = [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $session,
+        ];
+
+        // Test 1: Simple invalid query method (contains is not allowed)
+        $response = $this->receiveSubscribeMessageResponse(['documents'], $headers, [
+            Query::contains('status', ['active'])->toString(),
+        ]);
+        $this->assertEquals('error', $response['type']);
+        $this->assertStringContainsString('not supported in Realtime queries', $response['data']['message']);
+        $this->assertStringContainsString('contains', $response['data']['message']);
+
+        // Test 2: Invalid query method in nested AND query
+        $response = $this->receiveSubscribeMessageResponse(['documents'], $headers, [
+            Query::and([
+                Query::equal('status', ['active']),
+                Query::search('name', 'test'),
+            ])->toString(),
+        ]);
+        $this->assertEquals('error', $response['type']);
+        $this->assertStringContainsString('not supported in Realtime queries', $response['data']['message']);
+        $this->assertStringContainsString('search', $response['data']['message']);
+
+        // Test 3: Invalid query method in nested OR query
+        $response = $this->receiveSubscribeMessageResponse(['documents'], $headers, [
+            Query::or([
+                Query::equal('status', ['active']),
+                Query::between('score', 0, 100),
+            ])->toString(),
+        ]);
+        $this->assertEquals('error', $response['type']);
+        $this->assertStringContainsString('not supported in Realtime queries', $response['data']['message']);
+        $this->assertStringContainsString('between', $response['data']['message']);
+
+        // Test 4: Deeply nested invalid query (AND -> OR -> invalid)
+        $response = $this->receiveSubscribeMessageResponse(['documents'], $headers, [
+            Query::and([
+                Query::equal('status', ['active']),
+                Query::or([
+                    Query::greaterThan('score', 50),
+                    Query::startsWith('name', 'test'),
+                ]),
+            ])->toString(),
+        ]);
+        $this->assertEquals('error', $response['type']);
+        $this->assertStringContainsString('not supported in Realtime queries', $response['data']['message']);
+        $this->assertStringContainsString('startsWith', $response['data']['message']);
+
+        // Test 5: Multiple invalid 'queries' in nested structure
+        $response = $this->receiveSubscribeMessageResponse(['documents'], $headers, [
+            Query::and([
+                Query::contains('tags', ['important']),
+                Query::or([
+                    Query::endsWith('email', '@example.com'),
+                    Query::equal('status', ['active']),
+                ]),
+            ])->toString(),
+        ]);
+        $this->assertEquals('error', $response['type']);
+        $this->assertStringContainsString('not supported in Realtime queries', $response['data']['message']);
+        $this->assertTrue(
+            \str_contains($response['data']['message'], 'contains') ||
+            \str_contains($response['data']['message'], 'endsWith')
+        );
+    }
+
+    public function testProjectChannelWithHeaderOnly(): void
+    {
+        $user = $this->getUser();
+        $session = $user['session'] ?? '';
+        $projectId = $this->getProject()['$id'];
+
+        $client = $this->getWebsocketWithCustomQuery(
+            [
+                'project' => $projectId,
+            ],
+            [
+                'origin' => 'http://localhost',
+                'cookie' => 'a_session_' . $projectId . '=' . $session,
+                'x-appwrite-project' => $projectId,
+            ]
+        );
+
+        $response = \json_decode($client->receive(), true);
+        $this->assertSame('connected', $response['type']);
+        $subscribeResponse = $this->sendSubscribeMessage($client, [[
+            'channels' => ['project'],
+            'queries' => [Query::select(['*'])->toString()],
+        ]]);
+        $this->assertCount(1, $subscribeResponse['data']['subscriptions']);
+        $this->assertSame(['project'], $subscribeResponse['data']['subscriptions'][0]['channels']);
+
+        $client->close();
+
+        $clientWithQuery = $this->getWebsocketWithCustomQuery(
+            [
+                'project' => $projectId,
+            ],
+            [
+                'origin' => 'http://localhost',
+                'cookie' => 'a_session_' . $projectId . '=' . $session,
+                'x-appwrite-project' => $projectId,
+            ]
+        );
+
+        $response = \json_decode($clientWithQuery->receive(), true);
+        $this->assertSame('connected', $response['type']);
+        $subscribeResponseWithQuery = $this->sendSubscribeMessage($clientWithQuery, [[
+            'channels' => ['project'],
+            'queries' => [Query::select(['*'])->toString()],
+        ]]);
+        $this->assertCount(1, $subscribeResponseWithQuery['data']['subscriptions']);
+        $this->assertSame(['project'], $subscribeResponseWithQuery['data']['subscriptions'][0]['channels']);
+
+        $clientWithQuery->close();
+    }
+
+    public function testQueryMessageFiltersEvents(): void
+    {
+        $user = $this->getUser();
+        $session = $user['session'] ?? '';
+        $userId = $user['$id'] ?? '';
+        $projectId = $this->getProject()['$id'];
+
+        // Setup database and collection
+        $database = $this->client->call(Client::METHOD_POST, '/databases', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), [
+            'databaseId' => ID::unique(),
+            'name' => 'Query Message Test DB',
+        ]);
+        $databaseId = $database['body']['$id'];
+
+        $collection = $this->client->call(Client::METHOD_POST, '/databases/' . $databaseId . '/collections', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), [
+            'collectionId' => ID::unique(),
+            'name' => 'Query Message Test Collection',
+            'permissions' => [
+                Permission::create(Role::user($userId)),
+            ],
+            'documentSecurity' => true,
+        ]);
+        $collectionId = $collection['body']['$id'];
+
+        $this->client->call(Client::METHOD_POST, '/databases/' . $databaseId . '/collections/' . $collectionId . '/attributes/string', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), [
+            'key' => 'status',
+            'size' => 256,
+            'required' => false,
+        ]);
+
+        $this->assertEventually(function () use ($databaseId, $collectionId, $projectId) {
+            $response = $this->client->call(Client::METHOD_GET, '/databases/' . $databaseId . '/collections/' . $collectionId . '/attributes/status', \array_merge([
+                'content-type' => 'application/json',
+                'x-appwrite-project' => $projectId,
+                'x-appwrite-key' => $this->getProject()['apiKey'],
+            ]));
+            $this->assertEquals('available', $response['body']['status']);
+        }, 30000, 250);
+
+        $targetDocumentId = ID::unique();
+        $otherDocumentId = ID::unique();
+
+        $client = $this->getWebsocket(['documents'], [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $session,
+        ], null, [
+            Query::equal('$id', [$targetDocumentId])->toString(),
+        ]);
+
+        // Create matching document - should receive event
+        $this->client->call(Client::METHOD_POST, '/databases/' . $databaseId . '/collections/' . $collectionId . '/documents', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ], $this->getHeaders()), [
+            'documentId' => $targetDocumentId,
+            'data' => [
+                'status' => 'active',
+            ],
+            'permissions' => [
+                Permission::read(Role::any()),
+            ],
+        ]);
+
+        $event = \json_decode($client->receive(), true);
+        $this->assertEquals('event', $event['type']);
+        $this->assertEquals($targetDocumentId, $event['data']['payload']['$id']);
+
+        // Create non-matching document - should NOT receive event
+        $this->client->call(Client::METHOD_POST, '/databases/' . $databaseId . '/collections/' . $collectionId . '/documents', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ], $this->getHeaders()), [
+            'documentId' => $otherDocumentId,
+            'data' => [
+                'status' => 'inactive',
+            ],
+            'permissions' => [
+                Permission::read(Role::any()),
+            ],
+        ]);
+
+        try {
+            $client->receive();
+            $this->fail('Expected TimeoutException - event should be filtered by updated query');
+        } catch (TimeoutException $e) {
+            $this->assertTrue(true);
+        }
+
+        $client->close();
+    }
+}