From b7e2606b9f68dfd096cfec5cd18a82216273264e Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Mon, 22 Dec 2025 18:01:00 +0530 Subject: [PATCH] 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 | 20 +-- src/Appwrite/Messaging/Adapter/Realtime.php | 45 ++++++- .../Utopia/Database/Query/RuntimeQuery.php | 114 ++++++++++++++++++ tests/e2e/Services/Realtime/RealtimeBase.php | 4 +- .../Realtime/RealtimeCustomClientTest.php | 81 ++++++++++++- 5 files changed, 250 insertions(+), 14 deletions(-) create mode 100644 src/Appwrite/Utopia/Database/Query/RuntimeQuery.php diff --git a/app/realtime.php b/app/realtime.php index 31e6015d92..4bd105beb1 100644 --- a/app/realtime.php +++ b/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) { diff --git a/src/Appwrite/Messaging/Adapter/Realtime.php b/src/Appwrite/Messaging/Adapter/Realtime.php index 35b8089668..562be00e33 100644 --- a/src/Appwrite/Messaging/Adapter/Realtime.php +++ b/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. * diff --git a/src/Appwrite/Utopia/Database/Query/RuntimeQuery.php b/src/Appwrite/Utopia/Database/Query/RuntimeQuery.php new file mode 100644 index 0000000000..c887ca36d6 --- /dev/null +++ b/src/Appwrite/Utopia/Database/Query/RuntimeQuery.php @@ -0,0 +1,114 @@ + $queries + * @param array $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; + } +} diff --git a/tests/e2e/Services/Realtime/RealtimeBase.php b/tests/e2e/Services/Realtime/RealtimeBase.php index 89bd1898c4..ea5c3d710f 100644 --- a/tests/e2e/Services/Realtime/RealtimeBase.php +++ b/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( diff --git a/tests/e2e/Services/Realtime/RealtimeCustomClientTest.php b/tests/e2e/Services/Realtime/RealtimeCustomClientTest.php index c6a1686864..b15389dd2f 100644 --- a/tests/e2e/Services/Realtime/RealtimeCustomClientTest.php +++ b/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);