Enhance Realtime functionality with query support and improve tests

- Updated Realtime adapter to handle queries during subscription.
- Added query filtering capabilities in RuntimeQuery class.
- Modified RealtimeBase and RealtimeCustomClientTest to support query parameters in WebSocket connections.
- Improved test coverage for account and database channels with queries.

app/realtime.php

@@ -471,19 +471,20 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats,
 
                         $roles = $user->getRoles($database->getAuthorization());
                         $channels = $realtime->connections[$connection]['channels'];
+                        $queries = $realtime->connections[$connection]['queries'] ?? [];
 
                         $realtime->unsubscribe($connection);
-                        $realtime->subscribe($projectId, $connection, $roles, $channels);
+                        $realtime->subscribe($projectId, $connection, $roles, $channels, $queries);
                     }
                 }
 
                 $receivers = $realtime->getSubscribers($event);
 
-                if (App::isDevelopment() && !empty($receivers)) {
-                    Console::log("[Debug][Worker {$workerId}] Receivers: " . count($receivers));
-                    Console::log("[Debug][Worker {$workerId}] Receivers Connection IDs: " . json_encode($receivers));
-                    Console::log("[Debug][Worker {$workerId}] Event: " . $payload);
-                }
+                // if (App::isDevelopment() && !empty($receivers)) {
+                //     Console::log("[Debug][Worker {$workerId}] Receivers: " . count($receivers));
+                //     Console::log("[Debug][Worker {$workerId}] Receivers Connection IDs: " . json_encode($receivers));
+                //     Console::log("[Debug][Worker {$workerId}] Event: " . $payload);
+                // }
 
                 $server->send(
                     $receivers,
@@ -576,6 +577,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
         $roles = $user->getRoles($authorization);
 
         $channels = Realtime::convertChannels($request->getQuery('channels', []), $user->getId());
+        $queries = Realtime::convertQueries($request->getQuery('queries', []));
 
         /**
          * Channels Check
@@ -584,7 +586,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
             throw new Exception(Exception::REALTIME_POLICY_VIOLATION, 'Missing channels');
         }
 
-        $realtime->subscribe($project->getId(), $connection, $roles, $channels);
+        $realtime->subscribe($project->getId(), $connection, $roles, $channels, $queries);
 
         $realtime->connections[$connection]['authorization'] = $authorization;
 
@@ -594,6 +596,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
             'type' => 'connected',
             'data' => [
                 'channels' => array_keys($channels),
+                'queries' => array_keys($queries),
                 'user' => $user
             ]
         ]));
@@ -724,11 +727,12 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
 
                 $roles = $user->getRoles($database->getAuthorization());
                 $channels = Realtime::convertChannels(array_flip($realtime->connections[$connection]['channels']), $user->getId());
+                $queries = $realtime->connections[$connection]['queries'];
 
                 // Preserve authorization before subscribe overwrites the connection array
                 $authorization = $realtime->connections[$connection]['authorization'] ?? null;
 
-                $realtime->subscribe($realtime->connections[$connection]['projectId'], $connection, $roles, $channels);
+                $realtime->subscribe($realtime->connections[$connection]['projectId'], $connection, $roles, $channels, $queries);
 
                 // Restore authorization after subscribe
                 if ($authorization !== null) {

src/Appwrite/Messaging/Adapter/Realtime.php

@@ -2,12 +2,16 @@
 
 namespace Appwrite\Messaging\Adapter;
 
+use Appwrite\Extend\Exception;
 use Appwrite\Messaging\Adapter as MessagingAdapter;
 use Appwrite\PubSub\Adapter\Pool as PubSubPool;
+use Appwrite\Utopia\Database\Query\RuntimeQuery;
 use Utopia\Database\DateTime;
 use Utopia\Database\Document;
+use Utopia\Database\Exception\Query as QueryException;
 use Utopia\Database\Helpers\ID;
 use Utopia\Database\Helpers\Role;
+use Utopia\Database\Query;
 
 class Realtime extends MessagingAdapter
 {
@@ -51,9 +55,10 @@ class Realtime extends MessagingAdapter
      * @param mixed $identifier
      * @param array $roles
      * @param array $channels
+     * @param array $queries
      * @return void
      */
-    public function subscribe(string $projectId, mixed $identifier, array $roles, array $channels): void
+    public function subscribe(string $projectId, mixed $identifier, array $roles, array $channels, array $queries = []): void
     {
         if (!isset($this->subscriptions[$projectId])) { // Init Project
             $this->subscriptions[$projectId] = [];
@@ -72,7 +77,8 @@ class Realtime extends MessagingAdapter
         $this->connections[$identifier] = [
             'projectId' => $projectId,
             'roles' => $roles,
-            'channels' => $channels
+            'channels' => $channels,
+            'queries' => $queries
         ];
     }
 
@@ -206,7 +212,9 @@ class Realtime extends MessagingAdapter
                             /**
                              * To prevent duplicates, we save the connections as array keys.
                              */
-                            $receivers[$id] = 0;
+                            if (!empty(RuntimeQuery::filter($this->connections[$id]['queries'], $event['data']))) {
+                                $receivers[$id] = 0;
+                            }
                         }
                         break;
                     }
@@ -217,6 +225,19 @@ class Realtime extends MessagingAdapter
         return array_keys($receivers);
     }
 
+    public function filterEventData(array $documents, array $queries): array
+    {
+        if (empty($queries)) {
+            return $documents;
+        }
+        $filteredDocuments = [];
+        foreach ($documents as $document) {
+            $doc = new Document((array) $doc);
+        }
+
+        return $filteredDocuments;
+    }
+
     /**
      * Converts the channels from the Query Params into an array.
      * Also renames the account channel to account.USER_ID and removes all illegal account channel variations.
@@ -245,6 +266,24 @@ class Realtime extends MessagingAdapter
         return $channels;
     }
 
+    /**
+     * Converts the queries from the Query Params into an array.
+     * @param array $queries
+     * @return array
+     */
+    public static function convertQueries(array $queries): array
+    {
+        $queries = Query::parseQueries($queries);
+        foreach ($queries as $query) {
+            if (!in_array($query->getMethod(), RuntimeQuery::ALLOWED_QUERIES)) {
+                // TODO: add better error message with which queries are allowed
+                throw new QueryException(Exception::REALTIME_POLICY_VIOLATION, 'Query not supported');
+            }
+        }
+
+        return $queries;
+    }
+
     /**
      * Create channels array based on the event name and payload.
      *

src/Appwrite/Utopia/Database/Query/RuntimeQuery.php

@@ -0,0 +1,114 @@
+<?php
+
+namespace Appwrite\Utopia\Database\Query;
+
+use Utopia\Database\Query;
+
+class RuntimeQuery extends Query
+{
+    public const ALLOWED_QUERIES = [
+        // Equality & comparison
+        Query::TYPE_EQUAL,
+        Query::TYPE_NOT_EQUAL,
+        Query::TYPE_LESSER,
+        Query::TYPE_LESSER_EQUAL,
+        Query::TYPE_GREATER,
+        Query::TYPE_GREATER_EQUAL,
+
+        // Null checks
+        Query::TYPE_IS_NULL,
+        Query::TYPE_IS_NOT_NULL,
+
+        // Recursive checks
+        Query::TYPE_AND,
+        Query::TYPE_OR
+    ];
+
+    /**
+     * @param array<Query> $queries
+     * @param array<string, mixed> $payload
+     */
+    public static function filter(array $queries, array $payload): array
+    {
+        if (empty($queries)) {
+            return $payload;
+        }
+        foreach ($queries as $query) {
+            if (self::evaluateFilter($query, $payload)) {
+                return $payload;
+            };
+        }
+        return [];
+    }
+
+    private static function evaluateFilter(Query $query, array $payload): bool
+    {
+        $attribute = $query->getAttribute();
+        $method = $query->getMethod();
+        $values = $query->getValues();
+        if (!\array_key_exists($attribute, $payload)) {
+            return false;
+        }
+        $payloadAttributeValue = $payload[$attribute];
+        switch ($method) {
+            case Query::TYPE_EQUAL:
+                return self::anyMatch($values, fn ($value) => $payloadAttributeValue === $value);
+
+            case Query::TYPE_NOT_EQUAL:
+                return self::anyMatch($values, fn ($value) => $payloadAttributeValue !== $value);
+
+            case Query::TYPE_LESSER:
+                return self::anyMatch($values, fn ($value) => $payloadAttributeValue < $value);
+
+            case Query::TYPE_LESSER_EQUAL:
+                return self::anyMatch($values, fn ($value) => $payloadAttributeValue <= $value);
+
+            case Query::TYPE_GREATER:
+                return self::anyMatch($values, fn ($value) => $payloadAttributeValue > $value);
+
+            case Query::TYPE_GREATER_EQUAL:
+                return self::anyMatch($values, fn ($value) => $payloadAttributeValue >= $value);
+
+            case Query::TYPE_IS_NULL:
+                return $payloadAttributeValue === null;
+
+            case Query::TYPE_IS_NOT_NULL:
+                return $payloadAttributeValue !== null;
+
+            case Query::TYPE_AND:
+                foreach ($query->getValues() as $subquery) {
+                    // if any evaluation gets to false then whole and is false
+                    if (!self::evaluateFilter($subquery, $payload)) {
+                        return false;
+                    }
+                    return true;
+                }
+
+                // no break
+            case Query::TYPE_OR:
+                foreach ($query->getValues() as $subquery) {
+                    // if any evaluation gets to true then whole or is true
+                    if (self::evaluateFilter($subquery, $payload)) {
+                        return true;
+                    }
+                    return false;
+                }
+
+                // no break
+            default:
+                throw new \InvalidArgumentException(
+                    "Unsupported query method: {$method}"
+                );
+        }
+    }
+
+    private static function anyMatch(array $values, callable $fn): bool
+    {
+        foreach ($values as $value) {
+            if ($fn($value)) {
+                return true;
+            }
+        }
+        return false;
+    }
+}

tests/e2e/Services/Realtime/RealtimeBase.php

@@ -10,7 +10,8 @@ trait RealtimeBase
     private function getWebsocket(
         array $channels = [],
         array $headers = [],
-        string $projectId = null
+        string $projectId = null,
+        array $queries = []
     ): WebSocketClient {
         if (is_null($projectId)) {
             $projectId = $this->getProject()['$id'];
@@ -19,6 +20,7 @@ trait RealtimeBase
         $query = [
             "project" => $projectId,
             "channels" => $channels,
+            "queries" => $queries
         ];
 
         return new WebSocketClient(

tests/e2e/Services/Realtime/RealtimeCustomClientTest.php

@@ -12,6 +12,7 @@ use Tests\E2E\Services\Functions\FunctionsBase;
 use Utopia\Database\Helpers\ID;
 use Utopia\Database\Helpers\Permission;
 use Utopia\Database\Helpers\Role;
+use Utopia\Database\Query;
 use WebSocket\ConnectionException;
 use WebSocket\TimeoutException;
 
@@ -124,6 +125,82 @@ class RealtimeCustomClientTest extends Scope
         $client->close();
     }
 
+    public function testAccountChannelWithQueries()
+    {
+        $user = $this->getUser();
+        $userId = $user['$id'] ?? '';
+        $session = $user['session'] ?? '';
+        $projectId = $this->getProject()['$id'];
+
+        // Subscribe to account channel with a simple query
+        $client = $this->getWebsocket(['account'], [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $session,
+        ], null, [
+            Query::equal('$id', [$userId])->toString(),
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertArrayHasKey('type', $response);
+        $this->assertArrayHasKey('data', $response);
+        $this->assertEquals('connected', $response['type']);
+        $this->assertNotEmpty($response['data']);
+
+        // Channels still work as usual
+        $this->assertCount(2, $response['data']['channels']);
+        $this->assertContains('account', $response['data']['channels']);
+        $this->assertContains('account.' . $userId, $response['data']['channels']);
+
+        // Queries are echoed back in the connection payload
+        $this->assertArrayHasKey('queries', $response['data']);
+        $this->assertIsArray($response['data']['queries']);
+        $this->assertCount(1, $response['data']['queries']);
+
+        $this->assertNotEmpty($response['data']['user']);
+        $this->assertEquals($userId, $response['data']['user']['$id']);
+
+        $client->close();
+    }
+
+    public function testDatabaseChannelWithQueries()
+    {
+        $user = $this->getUser();
+        $session = $user['session'] ?? '';
+        $projectId = $this->getProject()['$id'];
+
+        // Subscribe to database-related channels with queries
+        $client = $this->getWebsocket(['documents', 'collections'], [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $session,
+        ], null, [
+            Query::equal('$id', ['dummy-id'])->toString(),
+            Query::isNotNull('payload')->toString(),
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertArrayHasKey('type', $response);
+        $this->assertArrayHasKey('data', $response);
+        $this->assertEquals('connected', $response['type']);
+        $this->assertNotEmpty($response['data']);
+
+        // Channels as in regular database test
+        $this->assertCount(2, $response['data']['channels']);
+        $this->assertContains('documents', $response['data']['channels']);
+        $this->assertContains('collections', $response['data']['channels']);
+
+        // Queries should be present
+        $this->assertArrayHasKey('queries', $response['data']);
+        $this->assertIsArray($response['data']['queries']);
+        $this->assertCount(2, $response['data']['queries']);
+
+        $this->assertNotEmpty($response['data']['user']);
+        $this->assertEquals($user['$id'], $response['data']['user']['$id']);
+
+        $client->close();
+    }
+
     public function testPingPong()
     {
         $client = $this->getWebsocket(['files'], [
@@ -692,8 +769,8 @@ class RealtimeCustomClientTest extends Scope
 
         $client = $this->getWebsocket(['documents', 'collections'], [
             'origin' => 'http://localhost',
-            'cookie' => 'a_session_' . $projectId . '=' . $session
-        ]);
+            'cookie' => 'a_session_' . $projectId . '=' . $session,
+        ], null);
 
         $response = json_decode($client->receive(), true);