added query per subscription and queryKeys along with the messages

app/realtime.php

@@ -481,25 +481,33 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats,
                     }
                 }
 
-                $receivers = $realtime->getSubscribers($event);
+                $receivers = $realtime->getSubscribers($event); // [connectionId => matchedQueryKeys[]]
 
                 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}] Receivers Connection IDs: " . json_encode(array_keys($receivers)));
                     Console::log("[Debug][Worker {$workerId}] Event: " . $payload);
                 }
 
-                $server->send(
-                    $receivers,
-                    json_encode([
-                        'type' => 'event',
-                        'data' => $event['data']
-                    ])
-                );
+                $totalMessages = 0;
 
-                if (($num = count($receivers)) > 0) {
-                    $register->get('telemetry.messageSentCounter')->add($num);
-                    $stats->incr($event['project'], 'messages', $num);
+                foreach ($receivers as $connectionId => $matchedQueryKeys) {
+                    $data = $event['data'];
+                    $data['queryKeys'] = $matchedQueryKeys;
+
+                    $server->send(
+                        [$connectionId],
+                        json_encode([
+                            'type' => 'event',
+                            'data' => $data
+                        ])
+                    );
+                    $totalMessages++;
+                }
+
+                if ($totalMessages > 0) {
+                    $register->get('telemetry.messageSentCounter')->add($totalMessages);
+                    $stats->incr($event['project'], 'messages', $totalMessages);
                 }
             });
         } catch (Throwable $th) {

src/Appwrite/Messaging/Adapter/Realtime.php

@@ -29,13 +29,19 @@ class Realtime extends MessagingAdapter
      *
      * [PROJECT_ID] ->
      *      [ROLE_X] ->
-     *          [CHANNEL_NAME_X] -> [CONNECTION_ID]
-     *          [CHANNEL_NAME_Y] -> [CONNECTION_ID]
-     *          [CHANNEL_NAME_Z] -> [CONNECTION_ID]
+     *          [CHANNEL_NAME_X] ->
+     *              [CONNECTION_ID] -> [QUERY_KEY] => true
+     *          [CHANNEL_NAME_Y] ->
+     *              [CONNECTION_ID] -> [QUERY_KEY] => true
+     *          [CHANNEL_NAME_Z] ->
+     *              [CONNECTION_ID] -> [QUERY_KEY] => true
      *      [ROLE_Y] ->
-     *          [CHANNEL_NAME_X] -> [CONNECTION_ID]
-     *          [CHANNEL_NAME_Y] -> [CONNECTION_ID]
-     *          [CHANNEL_NAME_Z] -> [CONNECTION_ID]
+     *          [CHANNEL_NAME_X] ->
+     *              [CONNECTION_ID] -> [QUERY_KEY] => true
+     *          [CHANNEL_NAME_Y] ->
+     *              [CONNECTION_ID] -> [QUERY_KEY] => true
+     *          [CHANNEL_NAME_Z] ->
+     *              [CONNECTION_ID] -> [QUERY_KEY] => true
      */
     public array $subscriptions = [];
 
@@ -63,21 +69,34 @@ class Realtime extends MessagingAdapter
             $this->subscriptions[$projectId] = [];
         }
 
+        $queryKeys = [];
+        if (empty($queries)) {
+            $queryKeys[] = '';
+        } else {
+            foreach ($queries as $query) {
+                /** @var Query $query */
+                $queryKeys[] = $query->toString();
+            }
+        }
+
         foreach ($roles as $role) {
             if (!isset($this->subscriptions[$projectId][$role])) { // Add user first connection
                 $this->subscriptions[$projectId][$role] = [];
             }
 
             foreach ($channels as $channel => $list) {
-                $this->subscriptions[$projectId][$role][$channel][$identifier] = true;
+                if (!isset($this->subscriptions[$projectId][$role][$channel][$identifier])) {
+                    $this->subscriptions[$projectId][$role][$channel][$identifier] = [];
+                }
+                foreach ($queryKeys as $queryKey) {
+                    $this->subscriptions[$projectId][$role][$channel][$identifier][$queryKey] = true;
+                }
             }
         }
-
         $this->connections[$identifier] = [
             'projectId' => $projectId,
             'roles' => $roles,
-            'channels' => $channels,
-            'queries' => $queries
+            'channels' => $channels
         ];
     }
 
@@ -91,10 +110,11 @@ class Realtime extends MessagingAdapter
     {
         $projectId = $this->connections[$connection]['projectId'] ?? '';
         $roles = $this->connections[$connection]['roles'] ?? [];
+        $channels = $this->connections[$connection]['channels'] ?? [];
 
         foreach ($roles as $role) {
-            foreach ($this->subscriptions[$projectId][$role] as $channel => $list) {
-                unset($this->subscriptions[$projectId][$role][$channel][$connection]); // Remove connection
+            foreach ($channels as $channel => $list) {
+                unset($this->subscriptions[$projectId][$role][$channel][$connection]); // dropping connection will drop the queries as well
 
                 if (empty($this->subscriptions[$projectId][$role][$channel])) {
                     unset($this->subscriptions[$projectId][$role][$channel]);  // Remove channel when no connections
@@ -130,7 +150,8 @@ class Realtime extends MessagingAdapter
 
         return array_key_exists($projectId, $this->subscriptions)
             && array_key_exists($role, $this->subscriptions[$projectId])
-            && array_key_exists($channel, $this->subscriptions[$projectId][$role]);
+            && array_key_exists($channel, $this->subscriptions[$projectId][$role])
+            && !empty($this->subscriptions[$projectId][$role][$channel]);
     }
 
     /**
@@ -207,18 +228,21 @@ class Realtime extends MessagingAdapter
                         /**
                          * Saving all connections that are allowed to receive this event.
                          */
-                        foreach (array_keys($this->subscriptions[$event['project']][$role][$channel]) as $id) {
-                            /**
-                             * To prevent duplicates, we save the connections as array keys.
-                             */
-                            $queries = $this->connections[$id]['queries'] ?? [];
-                            $payload = $event['data']['payload'] ?? [];
-                            if (
-                                empty($queries) ||
-                                !empty(RuntimeQuery::filter($queries, $payload))
-                            ) {
-                                $receivers[$id] = 0;
+                        $payload = $event['data']['payload'] ?? [];
+                        foreach ($this->subscriptions[$event['project']][$role][$channel] as $id => $queryMap) {
+                            $matchedQueryKeys = [];
+                            // for representing a all query subscribed channel
+                            if (isset($queryMap[''])) {
+                                $matchedQueryKeys[] = '';
+                            } else {
+                                foreach (array_keys($queryMap) as $queryKey) {
+                                    $parsed = Query::parseQueries([$queryKey]);
+                                    if (!empty(RuntimeQuery::filter($parsed, $payload))) {
+                                        $matchedQueryKeys[] = $queryKey;
+                                    }
+                                }
                             }
+                            $receivers[$id] = $matchedQueryKeys;
                         }
                         break;
                     }
@@ -226,7 +250,7 @@ class Realtime extends MessagingAdapter
             }
         }
 
-        return array_keys($receivers);
+        return $receivers;
     }
 
     /**

tests/e2e/Services/Realtime/RealtimeCustomClientQueryTest.php

@@ -1544,4 +1544,332 @@ class RealtimeCustomClientQueryTest extends Scope
             str_contains($response['data']['message'], 'endsWith')
         );
     }
+
+    public function testQueryKeys()
+    {
+        $user = $this->getUser();
+        $session = $user['session'] ?? '';
+        $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 Keys 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 Keys Collection',
+            'permissions' => [
+                Permission::create(Role::user($user['$id'])),
+            ],
+            'documentSecurity' => true,
+        ]);
+        $collectionId = $collection['body']['$id'];
+
+        // Attributes used by queries
+        $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,
+        ]);
+
+        sleep(2);
+
+        $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' => 'category',
+            'size' => 256,
+            'required' => false,
+        ]);
+
+        sleep(2);
+
+        $queryStatusActive = Query::equal('status', ['active'])->toString();
+        $queryStatusPending = Query::equal('status', ['pending'])->toString();
+        $queryComplex = Query::and([
+            Query::equal('status', ['active']),
+            Query::equal('category', ['gold']),
+        ])->toString();
+
+        // Subscribe with no queries -> should receive all events, queryKeys = ['']
+        $clientAll = $this->getWebsocket(['documents'], [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $session,
+        ]);
+
+        // Subscribe with query1 (status == active)
+        $clientQ1 = $this->getWebsocket(['documents'], [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $session,
+        ], null, [
+            $queryStatusActive,
+        ]);
+
+        // Subscribe with query2 (status == pending)
+        $clientQ2 = $this->getWebsocket(['documents'], [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $session,
+        ], null, [
+            $queryStatusPending,
+        ]);
+
+        // Subscribe with complex query (status == active AND category == gold)
+        $clientComplex = $this->getWebsocket(['documents'], [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $session,
+        ], null, [
+            $queryComplex,
+        ]);
+
+        // All clients should be connected
+        foreach ([$clientAll, $clientQ1, $clientQ2, $clientComplex] as $client) {
+            $response = json_decode($client->receive(), true);
+            $this->assertEquals('connected', $response['type']);
+        }
+
+        // 1) Create active/gold document -> should match Q1 and complex, and be seen by all
+        $docActiveGoldId = ID::unique();
+        $this->client->call(Client::METHOD_POST, '/databases/' . $databaseId . '/collections/' . $collectionId . '/documents', array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ], $this->getHeaders()), [
+            'documentId' => $docActiveGoldId,
+            'data' => [
+                'status' => 'active',
+                'category' => 'gold',
+            ],
+            'permissions' => [
+                Permission::read(Role::any()),
+            ],
+        ]);
+
+        // clientAll: should receive event, queryKeys = ['']
+        $eventAll = json_decode($clientAll->receive(), true);
+        $this->assertEquals('event', $eventAll['type']);
+        $this->assertEquals($docActiveGoldId, $eventAll['data']['payload']['$id']);
+        $this->assertArrayHasKey('queryKeys', $eventAll['data']);
+        $this->assertIsArray($eventAll['data']['queryKeys']);
+        $this->assertEquals([''], $eventAll['data']['queryKeys']);
+
+        // clientQ1: should receive event, queryKeys contains queryStatusActive
+        $eventQ1 = json_decode($clientQ1->receive(), true);
+        $this->assertEquals('event', $eventQ1['type']);
+        $this->assertEquals($docActiveGoldId, $eventQ1['data']['payload']['$id']);
+        $this->assertContains($queryStatusActive, $eventQ1['data']['queryKeys']);
+
+        // clientQ2: should NOT receive event (status is active, not pending)
+        try {
+            $clientQ2->receive();
+            $this->fail('Expected TimeoutException - event should be filtered for clientQ2 (active document)');
+        } catch (TimeoutException $e) {
+            $this->assertTrue(true);
+        }
+
+        // clientComplex: should receive event, queryKeys contains queryComplex
+        $eventComplex = json_decode($clientComplex->receive(), true);
+        $this->assertEquals('event', $eventComplex['type']);
+        $this->assertEquals($docActiveGoldId, $eventComplex['data']['payload']['$id']);
+        $this->assertContains($queryComplex, $eventComplex['data']['queryKeys']);
+
+        // 2) Create pending/silver document -> should match Q2 only, and be seen by all
+        $docPendingSilverId = ID::unique();
+        $this->client->call(Client::METHOD_POST, '/databases/' . $databaseId . '/collections/' . $collectionId . '/documents', array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ], $this->getHeaders()), [
+            'documentId' => $docPendingSilverId,
+            'data' => [
+                'status' => 'pending',
+                'category' => 'silver',
+            ],
+            'permissions' => [
+                Permission::read(Role::any()),
+            ],
+        ]);
+
+        // clientAll: should receive event, queryKeys = ['']
+        $eventAll2 = json_decode($clientAll->receive(), true);
+        $this->assertEquals('event', $eventAll2['type']);
+        $this->assertEquals($docPendingSilverId, $eventAll2['data']['payload']['$id']);
+        $this->assertArrayHasKey('queryKeys', $eventAll2['data']);
+        $this->assertIsArray($eventAll2['data']['queryKeys']);
+        $this->assertEquals([''], $eventAll2['data']['queryKeys']);
+
+        // clientQ1: should NOT receive event (status is pending)
+        try {
+            $clientQ1->receive();
+            $this->fail('Expected TimeoutException - event should be filtered for clientQ1 (pending document)');
+        } catch (TimeoutException $e) {
+            $this->assertTrue(true);
+        }
+
+        // clientQ2: should receive event, queryKeys contains queryStatusPending
+        $eventQ2 = json_decode($clientQ2->receive(), true);
+        $this->assertEquals('event', $eventQ2['type']);
+        $this->assertEquals($docPendingSilverId, $eventQ2['data']['payload']['$id']);
+        $this->assertContains($queryStatusPending, $eventQ2['data']['queryKeys']);
+
+        // clientComplex: should NOT receive event (status is pending, category silver)
+        try {
+            $clientComplex->receive();
+            $this->fail('Expected TimeoutException - event should be filtered for complex subscription (pending document)');
+        } catch (TimeoutException $e) {
+            $this->assertTrue(true);
+        }
+
+        $clientAll->close();
+        $clientQ1->close();
+        $clientQ2->close();
+        $clientComplex->close();
+    }
+
+    /**
+     * Ensure two separate subscriptions with different query keys
+     * only see their own matching events and expose the correct
+     * queryKey in queryKeys.
+     */
+    public function testMultipleSubscriptionsDifferentQueryKeys()
+    {
+        $user = $this->getUser();
+        $session = $user['session'] ?? '';
+        $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' => 'Multiple Query Keys 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' => 'Multiple Query Keys Collection',
+            'permissions' => [
+                Permission::create(Role::user($user['$id'])),
+            ],
+            'documentSecurity' => true,
+        ]);
+        $collectionId = $collection['body']['$id'];
+
+        // Attribute used by queries
+        $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,
+        ]);
+
+        sleep(2);
+
+        $queryStatusActive = Query::equal('status', ['active'])->toString();
+        $queryStatusPending = Query::equal('status', ['pending'])->toString();
+
+        // Two subscriptions on the same channel with different query keys
+        $clientQ1 = $this->getWebsocket(['documents'], [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $session,
+        ], null, [
+            $queryStatusActive,
+        ]);
+
+        $clientQ2 = $this->getWebsocket(['documents'], [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $session,
+        ], null, [
+            $queryStatusPending,
+        ]);
+
+        // Both should connect
+        $response = json_decode($clientQ1->receive(), true);
+        $this->assertEquals('connected', $response['type']);
+        $response = json_decode($clientQ2->receive(), true);
+        $this->assertEquals('connected', $response['type']);
+
+        // 1) active document -> only queryStatusActive subscription should see it
+        $docActiveId = ID::unique();
+        $this->client->call(Client::METHOD_POST, '/databases/' . $databaseId . '/collections/' . $collectionId . '/documents', array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ], $this->getHeaders()), [
+            'documentId' => $docActiveId,
+            'data' => [
+                'status' => 'active',
+            ],
+            'permissions' => [
+                Permission::read(Role::any()),
+            ],
+        ]);
+
+        $eventQ1 = json_decode($clientQ1->receive(), true);
+        $this->assertEquals('event', $eventQ1['type']);
+        $this->assertEquals($docActiveId, $eventQ1['data']['payload']['$id']);
+        $this->assertArrayHasKey('queryKeys', $eventQ1['data']);
+        $this->assertContains($queryStatusActive, $eventQ1['data']['queryKeys']);
+
+        try {
+            $clientQ2->receive();
+            $this->fail('Expected TimeoutException - clientQ2 should not receive active document');
+        } catch (TimeoutException $e) {
+            $this->assertTrue(true);
+        }
+
+        // 2) pending document -> only queryStatusPending subscription should see it
+        $docPendingId = ID::unique();
+        $this->client->call(Client::METHOD_POST, '/databases/' . $databaseId . '/collections/' . $collectionId . '/documents', array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ], $this->getHeaders()), [
+            'documentId' => $docPendingId,
+            'data' => [
+                'status' => 'pending',
+            ],
+            'permissions' => [
+                Permission::read(Role::any()),
+            ],
+        ]);
+
+        $eventQ2 = json_decode($clientQ2->receive(), true);
+        $this->assertEquals('event', $eventQ2['type']);
+        $this->assertEquals($docPendingId, $eventQ2['data']['payload']['$id']);
+        $this->assertArrayHasKey('queryKeys', $eventQ2['data']);
+        $this->assertContains($queryStatusPending, $eventQ2['data']['queryKeys']);
+
+        try {
+            $clientQ1->receive();
+            $this->fail('Expected TimeoutException - clientQ1 should not receive pending document');
+        } catch (TimeoutException $e) {
+            $this->assertTrue(true);
+        }
+
+        $clientQ1->close();
+        $clientQ2->close();
+    }
 }