Refactor realtime authorization handling and enhance presence event tests. Updated authorization roles synchronization in the realtime connection and added exception handling for user retrieval. Improved connection pool size logic for PubSub workers and added comprehensive tests for presence events to ensure correct message ordering and validation.

app/init/registers.php

@@ -333,7 +333,13 @@ $register->set('pools', function () {
 
             $poolAdapter = System::getEnv('_APP_POOL_ADAPTER', default: 'stack') === 'swoole' ? new SwoolePool() : new StackPool();
 
-            $pool = new Pool($poolAdapter, $name, $poolSize, function () use ($type, $resource, $dsn) {
+            // PubSub workers hold one long-lived subscribed connection and also need
+            // spare capacity for publishes from the same process.
+            $connectionPoolSize = $type === 'pubsub'
+                ? max(2, $poolSize)
+                : $poolSize;
+
+            $pool = new Pool($poolAdapter, $name, $connectionPoolSize, function () use ($type, $resource, $dsn) {
                 // Get Adapter
                 switch ($type) {
                     case 'database':

app/realtime.php

@@ -1,9 +1,9 @@
 <?php
 
-use Appwrite\Extend\Exception;
-use Appwrite\Extend\Exception as AppwriteException;
 use Appwrite\Event\Event as QueueEvent;
 use Appwrite\Event\Realtime as QueueRealtime;
+use Appwrite\Extend\Exception;
+use Appwrite\Extend\Exception as AppwriteException;
 use Appwrite\Messaging\Adapter\Realtime;
 use Appwrite\Network\Validator\Origin;
 use Appwrite\PubSub\Adapter\Pool as PubSubPool;
@@ -928,6 +928,17 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
             $authorization = new Authorization();
         }
 
+        // Ensure `$authorization` contains the same roles as the realtime connection.
+        // `setPermission()` validates against `$authorization->getRoles()`, but roles are
+        // computed/stored separately in the realtime adapter connection tree.
+        $connectionRoles = $realtime->connections[$connection]['roles'] ?? [];
+        foreach ($connectionRoles as $role) {
+            if ($authorization->hasRole($role)) {
+                continue;
+            }
+            $authorization->addRole($role);
+        }
+
         $database = getConsoleDB();
         $database->setAuthorization($authorization);
 
@@ -1178,6 +1189,10 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
                 /** @var User $user */
                 $user = $database->getDocument('users', $userId);
 
+                if ($user->isEmpty()) {
+                    throw new Exception(Exception::USER_NOT_FOUND, params:[$userId]);
+                }
+
                 if (!is_array($message['data'])) {
                     throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Payload is not valid.');
                 }

tests/e2e/Services/Presence/PresenceRealtimeClientTest.php

@@ -12,6 +12,7 @@ use Utopia\Database\Helpers\Permission;
 use Utopia\Database\Helpers\Role;
 use Utopia\Database\Query;
 use WebSocket\Client as WebSocketClient;
+use WebSocket\TimeoutException;
 
 class PresenceRealtimeClientTest extends Scope
 {
@@ -19,6 +20,130 @@ class PresenceRealtimeClientTest extends Scope
     use RealtimeBase;
     use SideClient;
 
+    private function assertPresenceRealtimeEvent(
+        array $event,
+        string $presenceId,
+        string $action,
+        string $status,
+        array $metadata = [],
+        ?string $expectedUserId = null
+    ): void {
+        $expectedUserId ??= $this->getUser()['$id'];
+        $this->assertSame('event', $event['type'] ?? null);
+        $this->assertContains('presences', $event['data']['channels'] ?? []);
+        $this->assertContains('presences.' . $presenceId, $event['data']['channels'] ?? []);
+        $this->assertNotEmpty($event['data']['events'] ?? []);
+        $this->assertContains('presences.' . $presenceId . '.' . $action, $event['data']['events'] ?? []);
+        $this->assertNotEmpty($event['data']['timestamp'] ?? null);
+        $this->assertArrayHasKey('subscriptions', $event['data'] ?? []);
+        $this->assertNotEmpty($event['data']['subscriptions'] ?? []);
+        $this->assertSame($presenceId, $event['data']['payload']['$id'] ?? null);
+        $this->assertSame($status, $event['data']['payload']['status'] ?? null);
+        $this->assertSame($metadata, $event['data']['payload']['metadata'] ?? []);
+        $this->assertSame($expectedUserId, $event['data']['payload']['userId'] ?? null);
+    }
+
+    private function assertNoRealtimeEvent(WebSocketClient $client): void
+    {
+        try {
+            $client->receive();
+            $this->fail('Expected TimeoutException - event should not be received');
+        } catch (TimeoutException $e) {
+            $this->assertTrue(true);
+        }
+    }
+
+    /**
+     * Presence websocket contract: after sending a `type: presence` message,
+     * the sender socket should receive:
+     * 1) `type: response` (for the persistence write)
+     * 2) `type: event` (for the realtime upsert)
+     *
+     * This keeps the tests strict about ordering and avoids leaving unread
+     * realtime events in the socket buffer for later assertions.
+     */
+    private function assertPresenceResponseThenUpsertEvent(
+        WebSocketClient $client,
+        string $expectedStatus,
+        array $expectedMetadata,
+        ?string $expectedUserId = null
+    ): string {
+        $expectedUserId ??= $this->getUser()['$id'];
+
+        $presenceId = null;
+        $response = null;
+        $event = null;
+
+        // Ordering is not guaranteed because:
+        // - response is sent directly via $server->send(...)
+        // - event is emitted via pub/sub and may arrive earlier
+        for ($attempts = 0; $attempts < 5; $attempts++) {
+            $message = \json_decode($client->receive(), true);
+            $type = $message['type'] ?? null;
+
+            if ($type === 'response') {
+                $response ??= $message;
+
+                $this->assertSame('presence', $response['data']['to'] ?? null);
+                $presenceId ??= $response['data']['presence']['$id'] ?? null;
+                $this->assertNotEmpty($presenceId);
+                $this->assertSame($expectedStatus, $response['data']['presence']['status'] ?? null);
+                $this->assertSame($expectedMetadata, $response['data']['presence']['metadata'] ?? null);
+            } elseif ($type === 'event') {
+                $event ??= $message;
+                $presenceId ??= $event['data']['payload']['$id'] ?? null;
+                $this->assertNotEmpty($presenceId);
+
+                $this->assertPresenceRealtimeEvent(
+                    $event,
+                    $presenceId,
+                    'upsert',
+                    $expectedStatus,
+                    $expectedMetadata,
+                    $expectedUserId
+                );
+            }
+
+            if ($response !== null && $event !== null) {
+                return $presenceId;
+            }
+        }
+
+        $this->fail('Expected both realtime presence `response` and `event` messages');
+        return '';
+    }
+
+    /**
+     * After getting a `response` for a presence message, the next interesting
+     * realtime message should be the corresponding `event` for the same presence id.
+     */
+    private function receivePresenceEvent(
+        WebSocketClient $client,
+        string $presenceId,
+        string $action,
+        string $status,
+        array $expectedMetadata,
+        ?string $expectedUserId = null
+    ): array {
+        do {
+            $message = \json_decode($client->receive(), true);
+        } while (
+            ($message['type'] ?? null) !== 'event'
+            || ($message['data']['payload']['$id'] ?? null) !== $presenceId
+        );
+
+        $this->assertPresenceRealtimeEvent(
+            $message,
+            $presenceId,
+            $action,
+            $status,
+            $expectedMetadata,
+            $expectedUserId
+        );
+
+        return $message;
+    }
+
     private function connectPresenceSocket(bool $authenticated = true, int $timeout = 2): WebSocketClient
     {
         $headers = [
@@ -61,7 +186,7 @@ class PresenceRealtimeClientTest extends Scope
     {
         $presenceId = ID::unique();
         $userId = $this->getUser()['$id'];
-        $client = $this->connectPresenceSocket();
+        $client = $this->connectPresenceSocket(true, 5);
 
         $client->send(\json_encode([
             'type' => 'presence',
@@ -75,13 +200,12 @@ class PresenceRealtimeClientTest extends Scope
             ],
         ]));
 
-        $response = \json_decode($client->receive(), true);
-        $this->assertSame('response', $response['type'] ?? null);
-        $this->assertSame('presence', $response['data']['to'] ?? null);
-        $this->assertSame($presenceId, $response['data']['presence']['$id'] ?? null);
-        $this->assertSame($userId, $response['data']['presence']['userId'] ?? null);
-        $this->assertSame('online', $response['data']['presence']['status'] ?? null);
-        $this->assertSame(['device' => 'web'], $response['data']['presence']['metadata'] ?? null);
+        $this->assertPresenceResponseThenUpsertEvent(
+            $client,
+            'online',
+            ['device' => 'web'],
+            $userId
+        );
 
         $read = $this->client->call(
             Client::METHOD_GET,
@@ -102,7 +226,7 @@ class PresenceRealtimeClientTest extends Scope
     {
         $presenceId = ID::unique();
         $userId = $this->getUser()['$id'];
-        $client = $this->connectPresenceSocket();
+        $client = $this->connectPresenceSocket(true, 5);
 
         $client->send(\json_encode([
             'type' => 'presence',
@@ -115,10 +239,12 @@ class PresenceRealtimeClientTest extends Scope
                 'permissions' => $this->getPresencePermissions($userId),
             ],
         ]));
-        $first = \json_decode($client->receive(), true);
-        $this->assertSame('response', $first['type'] ?? null);
-        $this->assertSame($presenceId, $first['data']['presence']['$id'] ?? null);
-        $this->assertSame('away', $first['data']['presence']['status'] ?? null);
+        $this->assertPresenceResponseThenUpsertEvent(
+            $client,
+            'away',
+            ['source' => 'first'],
+            $userId
+        );
 
         $client->send(\json_encode([
             'type' => 'presence',
@@ -131,11 +257,12 @@ class PresenceRealtimeClientTest extends Scope
                 'permissions' => $this->getPresencePermissions($userId),
             ],
         ]));
-        $second = \json_decode($client->receive(), true);
-        $this->assertSame('response', $second['type'] ?? null);
-        $this->assertSame($presenceId, $second['data']['presence']['$id'] ?? null);
-        $this->assertSame('busy', $second['data']['presence']['status'] ?? null);
-        $this->assertSame(['source' => 'second'], $second['data']['presence']['metadata'] ?? null);
+        $this->assertPresenceResponseThenUpsertEvent(
+            $client,
+            'busy',
+            ['source' => 'second'],
+            $userId
+        );
 
         $list = $this->client->call(
             Client::METHOD_GET,
@@ -160,6 +287,70 @@ class PresenceRealtimeClientTest extends Scope
         $client->close();
     }
 
+    public function testPresenceMessageUpsertWithSameUserPersistsSingleRecord(): void
+    {
+        $firstPresenceId = ID::unique();
+        $secondPresenceId = ID::unique();
+        $userId = $this->getUser()['$id'];
+        $client = $this->connectPresenceSocket(true, 5);
+
+        $client->send(\json_encode([
+            'type' => 'presence',
+            'data' => [
+                'presenceId' => $firstPresenceId,
+                'status' => 'away',
+                'metadata' => [
+                    'source' => 'first-user-upsert',
+                ],
+                'permissions' => $this->getPresencePermissions($userId),
+            ],
+        ]));
+        $this->assertPresenceResponseThenUpsertEvent(
+            $client,
+            'away',
+            ['source' => 'first-user-upsert'],
+            $userId
+        );
+
+        $client->send(\json_encode([
+            'type' => 'presence',
+            'data' => [
+                'presenceId' => $secondPresenceId,
+                'status' => 'busy',
+                'metadata' => [
+                    'source' => 'second-user-upsert',
+                ],
+                'permissions' => $this->getPresencePermissions($userId),
+            ],
+        ]));
+        $this->assertPresenceResponseThenUpsertEvent(
+            $client,
+            'busy',
+            ['source' => 'second-user-upsert'],
+            $userId
+        );
+
+        $list = $this->client->call(
+            Client::METHOD_GET,
+            '/presences',
+            $this->getServerHeaders(),
+            [
+                'queries' => [
+                    Query::equal('userId', [$userId])->toString(),
+                ],
+            ]
+        );
+
+        $this->assertSame(200, $list['headers']['status-code']);
+        $this->assertSame(1, $list['body']['total']);
+        $this->assertCount(1, $list['body']['presences']);
+        $this->assertSame($userId, $list['body']['presences'][0]['userId']);
+        $this->assertSame('busy', $list['body']['presences'][0]['status']);
+        $this->assertSame(['source' => 'second-user-upsert'], $list['body']['presences'][0]['metadata']);
+
+        $client->close();
+    }
+
     public function testPresenceMessageValidationErrors(): void
     {
         $client = $this->connectPresenceSocket();
@@ -240,13 +431,13 @@ class PresenceRealtimeClientTest extends Scope
         $this->assertSame(200, $create['headers']['status-code']);
 
         $createEvent = \json_decode($client->receive(), true);
-        $this->assertSame('event', $createEvent['type'] ?? null);
-        $this->assertContains('presences', $createEvent['data']['channels'] ?? []);
-        $this->assertContains('presences.' . $presenceId, $createEvent['data']['channels'] ?? []);
-        $this->assertNotEmpty($createEvent['data']['events'] ?? []);
-        $this->assertContains('presences.' . $presenceId . '.upsert', $createEvent['data']['events'] ?? []);
-        $this->assertSame($presenceId, $createEvent['data']['payload']['$id'] ?? null);
-        $this->assertSame('online', $createEvent['data']['payload']['status'] ?? null);
+        $this->assertPresenceRealtimeEvent(
+            $createEvent,
+            $presenceId,
+            'upsert',
+            'online',
+            ['source' => 'channel-parsing-create']
+        );
 
         $update = $this->client->call(
             Client::METHOD_PATCH,
@@ -260,13 +451,13 @@ class PresenceRealtimeClientTest extends Scope
         $this->assertSame(200, $update['headers']['status-code']);
 
         $updateEvent = \json_decode($client->receive(), true);
-        $this->assertSame('event', $updateEvent['type'] ?? null);
-        $this->assertContains('presences', $updateEvent['data']['channels'] ?? []);
-        $this->assertContains('presences.' . $presenceId, $updateEvent['data']['channels'] ?? []);
-        $this->assertNotEmpty($updateEvent['data']['events'] ?? []);
-        $this->assertContains('presences.' . $presenceId . '.update', $updateEvent['data']['events'] ?? []);
-        $this->assertSame($presenceId, $updateEvent['data']['payload']['$id'] ?? null);
-        $this->assertSame('away', $updateEvent['data']['payload']['status'] ?? null);
+        $this->assertPresenceRealtimeEvent(
+            $updateEvent,
+            $presenceId,
+            'update',
+            'away',
+            ['source' => 'channel-parsing-update']
+        );
 
         $delete = $this->client->call(
             Client::METHOD_DELETE,
@@ -276,12 +467,13 @@ class PresenceRealtimeClientTest extends Scope
         $this->assertSame(204, $delete['headers']['status-code']);
 
         $deleteEvent = \json_decode($client->receive(), true);
-        $this->assertSame('event', $deleteEvent['type'] ?? null);
-        $this->assertContains('presences', $deleteEvent['data']['channels'] ?? []);
-        $this->assertContains('presences.' . $presenceId, $deleteEvent['data']['channels'] ?? []);
-        $this->assertNotEmpty($deleteEvent['data']['events'] ?? []);
-        $this->assertContains('presences.' . $presenceId . '.delete', $deleteEvent['data']['events'] ?? []);
-        $this->assertSame($presenceId, $deleteEvent['data']['payload']['$id'] ?? null);
+        $this->assertPresenceRealtimeEvent(
+            $deleteEvent,
+            $presenceId,
+            'delete',
+            'away',
+            ['source' => 'channel-parsing-update']
+        );
 
         $client->close();
     }
@@ -298,6 +490,11 @@ class PresenceRealtimeClientTest extends Scope
         $listener = $this->getWebsocket(['presences', 'presences.' . $presenceId], $headers, timeout: 8);
         $connected = \json_decode($listener->receive(), true);
         $this->assertSame('connected', $connected['type'] ?? null);
+        $this->assertCount(2, $connected['data']['channels'] ?? []);
+        $this->assertContains('presences', $connected['data']['channels'] ?? []);
+        $this->assertContains('presences.' . $presenceId, $connected['data']['channels'] ?? []);
+        $this->assertCount(1, $connected['data']['subscriptions'] ?? []);
+        $this->assertNotEmpty(array_values($connected['data']['subscriptions'] ?? []));
 
         $publisher = $this->connectPresenceSocket(true, timeout: 8);
 
@@ -313,24 +510,211 @@ class PresenceRealtimeClientTest extends Scope
             ],
         ]));
 
-        $createResponse = \json_decode($publisher->receive(), true);
-        $this->assertSame('response', $createResponse['type'] ?? null);
-        $this->assertSame('presence', $createResponse['data']['to'] ?? null);
-        $this->assertSame($presenceId, $createResponse['data']['presence']['$id'] ?? null);
+        $receivedPresenceId = $this->assertPresenceResponseThenUpsertEvent(
+            $publisher,
+            'online',
+            ['source' => 'realtime-create-delete-events'],
+            $userId
+        );
+        $this->assertSame($presenceId, $receivedPresenceId);
 
         $createEvent = \json_decode($listener->receive(), true);
-        $this->assertSame('event', $createEvent['type'] ?? null);
-        $this->assertContains('presences.' . $presenceId . '.upsert', $createEvent['data']['events'] ?? []);
-        $this->assertSame($presenceId, $createEvent['data']['payload']['$id'] ?? null);
-        $this->assertSame('online', $createEvent['data']['payload']['status'] ?? null);
+        $this->assertPresenceRealtimeEvent(
+            $createEvent,
+            $presenceId,
+            'upsert',
+            'online',
+            ['source' => 'realtime-create-delete-events']
+        );
 
         $publisher->close();
 
         $deleteEvent = \json_decode($listener->receive(), true);
-        $this->assertSame('event', $deleteEvent['type'] ?? null);
-        $this->assertContains('presences.' . $presenceId . '.delete', $deleteEvent['data']['events'] ?? []);
-        $this->assertSame($presenceId, $deleteEvent['data']['payload']['$id'] ?? null);
+        $this->assertPresenceRealtimeEvent(
+            $deleteEvent,
+            $presenceId,
+            'delete',
+            'online',
+            ['source' => 'realtime-create-delete-events']
+        );
 
         $listener->close();
     }
+
+    public function testPresencePermission(): void
+    {
+        $presenceIdAny = ID::unique();
+        $presenceIdUsers = ID::unique();
+        $presenceIdOwner = ID::unique();
+
+        $user1 = $this->getUser();
+        $user1Id = $user1['$id'];
+        $user2 = $this->getUser(true);
+        $user3 = $this->getUser(true);
+        $projectId = $this->getProject()['$id'];
+
+        $user1Headers = [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $user1['session'],
+        ];
+
+        $user2Headers = [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $user2['session'],
+        ];
+
+        $user3Headers = [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_' . $projectId . '=' . $user3['session'],
+        ];
+
+        $user1Listener = $this->getWebsocket(['presences', 'presences.' . $presenceIdAny, 'presences.' . $presenceIdUsers, 'presences.' . $presenceIdOwner], $user1Headers, timeout: 3);
+        $user2Listener = $this->getWebsocket(['presences', 'presences.' . $presenceIdAny, 'presences.' . $presenceIdUsers, 'presences.' . $presenceIdOwner], $user2Headers, timeout: 3);
+        $user3Listener = $this->getWebsocket(['presences', 'presences.' . $presenceIdAny, 'presences.' . $presenceIdUsers, 'presences.' . $presenceIdOwner], $user3Headers, timeout: 3);
+
+        $this->assertSame('connected', (\json_decode($user1Listener->receive(), true))['type'] ?? null);
+        $this->assertSame('connected', (\json_decode($user2Listener->receive(), true))['type'] ?? null);
+        $this->assertSame('connected', (\json_decode($user3Listener->receive(), true))['type'] ?? null);
+
+        $publisher = $this->getWebsocket(['presences'], $user1Headers, timeout: 5);
+        $this->assertSame('connected', (\json_decode($publisher->receive(), true))['type'] ?? null);
+
+        $publisher->send(\json_encode([
+            'type' => 'presence',
+            'data' => [
+                'presenceId' => $presenceIdAny,
+                'status' => 'online',
+                'metadata' => [
+                    'visibility' => 'any',
+                ],
+                'permissions' => [
+                    Permission::read(Role::any()),
+                    Permission::update(Role::any()),
+                    Permission::delete(Role::any()),
+                ],
+            ],
+        ]));
+
+        $receivedPresenceId = $this->assertPresenceResponseThenUpsertEvent(
+            $publisher,
+            'online',
+            ['visibility' => 'any'],
+            $user1Id
+        );
+        $this->assertSame($presenceIdAny, $receivedPresenceId);
+
+        $this->assertPresenceRealtimeEvent(
+            \json_decode($user1Listener->receive(), true),
+            $presenceIdAny,
+            'upsert',
+            'online',
+            ['visibility' => 'any'],
+            $user1Id
+        );
+        $this->assertPresenceRealtimeEvent(
+            \json_decode($user2Listener->receive(), true),
+            $presenceIdAny,
+            'upsert',
+            'online',
+            ['visibility' => 'any'],
+            $user1Id
+        );
+        $this->assertPresenceRealtimeEvent(
+            \json_decode($user3Listener->receive(), true),
+            $presenceIdAny,
+            'upsert',
+            'online',
+            ['visibility' => 'any'],
+            $user1Id
+        );
+
+        $publisher->send(\json_encode([
+            'type' => 'presence',
+            'data' => [
+                'presenceId' => $presenceIdUsers,
+                'status' => 'away',
+                'metadata' => [
+                    'visibility' => 'users',
+                ],
+                'permissions' => [
+                    Permission::read(Role::users()),
+                    Permission::update(Role::users()),
+                    Permission::delete(Role::users()),
+                ],
+            ],
+        ]));
+
+        $receivedPresenceId = $this->assertPresenceResponseThenUpsertEvent(
+            $publisher,
+            'away',
+            ['visibility' => 'users'],
+            $user1Id
+        );
+        $this->assertSame($presenceIdUsers, $receivedPresenceId);
+
+        $this->assertPresenceRealtimeEvent(
+            \json_decode($user1Listener->receive(), true),
+            $presenceIdUsers,
+            'upsert',
+            'away',
+            ['visibility' => 'users'],
+            $user1Id
+        );
+        $this->assertPresenceRealtimeEvent(
+            \json_decode($user3Listener->receive(), true),
+            $presenceIdUsers,
+            'upsert',
+            'away',
+            ['visibility' => 'users'],
+            $user1Id
+        );
+        $this->assertPresenceRealtimeEvent(
+            \json_decode($user2Listener->receive(), true),
+            $presenceIdUsers,
+            'upsert',
+            'away',
+            ['visibility' => 'users'],
+            $user1Id
+        );
+
+        $publisher->send(\json_encode([
+            'type' => 'presence',
+            'data' => [
+                'presenceId' => $presenceIdOwner,
+                'status' => 'busy',
+                'metadata' => [
+                    'visibility' => 'owner',
+                ],
+                'permissions' => [
+                    Permission::read(Role::user($user1Id)),
+                    Permission::update(Role::user($user1Id)),
+                    Permission::delete(Role::user($user1Id)),
+                ],
+            ],
+        ]));
+
+        $receivedPresenceId = $this->assertPresenceResponseThenUpsertEvent(
+            $publisher,
+            'busy',
+            ['visibility' => 'owner'],
+            $user1Id
+        );
+        $this->assertSame($presenceIdOwner, $receivedPresenceId);
+
+        $this->assertPresenceRealtimeEvent(
+            \json_decode($user1Listener->receive(), true),
+            $presenceIdOwner,
+            'upsert',
+            'busy',
+            ['visibility' => 'owner'],
+            $user1Id
+        );
+        $this->assertNoRealtimeEvent($user2Listener);
+        $this->assertNoRealtimeEvent($user3Listener);
+
+        $publisher->close();
+        $user1Listener->close();
+        $user2Listener->close();
+        $user3Listener->close();
+    }
 }