Merge branch 'feat-265-realtime' into feat-265-realtime-resources

app/controllers/api/health.php

@@ -33,17 +33,6 @@ App::get('/v1/health/version')
         $response->json(['version' => APP_VERSION_STABLE]);
     });
 
-App::get('/v1/health/realtime')
-    ->desc('Get Realtime')
-    ->groups(['api', 'health'])
-    ->label('scope', 'public')
-    ->inject('response')
-    ->action(function ($response) {
-        /** @var Utopia\Response $response */
-        // TODO: realtime health
-        $response->json(['status' => 'OK']);
-    });
-
 App::get('/v1/health/db')
     ->desc('Get DB')
     ->groups(['api', 'health'])

app/realtime.php

@@ -128,12 +128,12 @@ $server->on('start', function (Server $server) {
 });
 
 $server->on('open', function (Server $server, Request $request) use (&$connections, &$subscriptions, &$register) {
-    Console::info("Connection open (user: {$request->fd}, connections: {}, worker: {$server->getWorkerId()})");
-
     $app = new App('');
     $connection = $request->fd;
     $request = new SwooleRequest($request);
 
+    Console::info("Connection open (user: {$connection}, worker: {$server->getWorkerId()})");
+
     App::setResource('request', function () use ($request) {
         return $request;
     });
@@ -151,65 +151,68 @@ $server->on('open', function (Server $server, Request $request) use (&$connectio
     /** @var Appwrite\Database\Document $console */
     $console = $app->getResource('console');
 
-    /*
-     *  Project Check
-     */
-    if (empty($project->getId())) {
-        $server->push($connection, 'Missing or unknown project ID');
+    try {
+        /*
+        *  Project Check
+        */
+        if (empty($project->getId())) {
+            throw new Exception('Missing or unknown project ID', 1008);
+        }
+
+        /*
+        * Abuse Check
+        *
+        * Abuse limits are connecting 128 times per minute and ip address.
+        */
+        $timeLimit = new TimeLimit('url:{url},ip:{ip}', 128, 60, function () use ($register) {
+            return $register->get('db');
+        });
+        $timeLimit
+            ->setNamespace('app_' . $project->getId())
+            ->setParam('{ip}', $request->getIP())
+            ->setParam('{url}', $request->getURI());
+
+        $abuse = new Abuse($timeLimit);
+
+        if ($abuse->check() && App::getEnv('_APP_OPTIONS_ABUSE', 'enabled') === 'enabled') {
+            throw new Exception('Too many requests', 1013);
+        }
+
+        /*
+        * Validate Client Domain - Check to avoid CSRF attack.
+        * Adding Appwrite API domains to allow XDOMAIN communication.
+        * Skip this check for non-web platforms which are not required to send an origin header.
+        */
+        $origin = $request->getOrigin();
+        $originValidator = new Origin(\array_merge($project->getAttribute('platforms', []), $console->getAttribute('platforms', [])));
+
+        if (!$originValidator->isValid($origin)) {
+            throw new Exception($originValidator->getDescription(), 1008);
+        }
+
+        Realtime::setUser($user);
+
+        $roles = Realtime::getRoles();
+        $channels = Realtime::parseChannels($request->getQuery('channels', []));
+
+        /**
+         * Channels Check
+         */
+        if (empty($channels)) {
+            throw new Exception('Missing channels', 1008);
+        }
+
+        Realtime::subscribe($project->getId(), $connection, $roles, $subscriptions, $connections, $channels);
+
+        $server->push($connection, json_encode($channels));
+    } catch (\Throwable $th) {
+        $response = [
+            'code' => $th->getCode(),
+            'message' => $th->getMessage()
+        ];
+        $server->push($connection, json_encode($response));
         $server->close($connection);
-        return;
     }
-
-    /*
-     * Abuse Check
-     */
-    $timeLimit = new TimeLimit('url:{url},ip:{ip}', 60, 60, function () use ($register) {
-        return $register->get('db');
-    });
-    $timeLimit
-        ->setNamespace('app_' . $project->getId())
-        ->setParam('{ip}', $request->getIP())
-        ->setParam('{url}', $request->getURI());
-
-    $abuse = new Abuse($timeLimit);
-
-    if ($abuse->check() && App::getEnv('_APP_OPTIONS_ABUSE', 'enabled') === 'enabled') {
-        $server->push($connection, 'Too many requests');
-        $server->close($connection);
-        return;
-    }
-
-    /*
-     * Validate Client Domain - Check to avoid CSRF attack.
-     * Adding Appwrite API domains to allow XDOMAIN communication.
-     * Skip this check for non-web platforms which are not required to send an origin header.
-     */
-    $origin = $request->getOrigin();
-    $originValidator = new Origin(\array_merge($project->getAttribute('platforms', []), $console->getAttribute('platforms', [])));
-
-    if (!$originValidator->isValid($origin)) {
-        $server->push($connection, $originValidator->getDescription());
-        $server->close($connection);
-        return;
-    }
-
-    Realtime::setUser($user);
-
-    $roles = Realtime::getRoles();
-    $channels = Realtime::parseChannels($request->getQuery('channels', []));
-
-    /**
-     * Channels Check
-     */
-    if (empty($channels)) {
-        $server->push($connection, 'Missing channels');
-        $server->close($connection);
-        return;
-    }
-
-    Realtime::subscribe($project->getId(), $connection, $roles, $subscriptions, $connections, $channels);
-
-    $server->push($connection, json_encode($channels));
 });
 
 $server->on('message', function (Server $server, Frame $frame) {
@@ -217,9 +220,9 @@ $server->on('message', function (Server $server, Frame $frame) {
     $server->close($frame->fd);
 });
 
-$server->on('close', function (Server $server, int $fd) use (&$connections, &$subscriptions) {
-    Realtime::unsubscribe($fd, $subscriptions, $connections);
-    Console::info('Connection close: ' . $fd);
+$server->on('close', function (Server $server, int $connection) use (&$connections, &$subscriptions) {
+    Realtime::unsubscribe($connection, $subscriptions, $connections);
+    Console::info('Connection close: ' . $connection);
 });
 
 $server->start();

composer.json

@@ -63,6 +63,7 @@
         "appwrite/sdk-generator": "0.5.5",
         "phpunit/phpunit": "9.4.2",
         "swoole/ide-helper": "4.5.5",
+        "textalk/websocket": "1.5.2",
         "vimeo/psalm": "4.1.1"
     },
     "repositories": [

composer.lock

@@ -5027,6 +5027,86 @@
             ],
             "time": "2021-03-08T21:52:55+00:00"
         },
+        {
+            "name": "symfony/polyfill-ctype",
+            "version": "dev-main",
+            "source": {
+                "type": "git",
+                "url": "https://github.com/symfony/polyfill-ctype.git",
+                "reference": "c6c942b1ac76c82448322025e084cadc56048b4e"
+            },
+            "dist": {
+                "type": "zip",
+                "url": "https://api.github.com/repos/symfony/polyfill-ctype/zipball/c6c942b1ac76c82448322025e084cadc56048b4e",
+                "reference": "c6c942b1ac76c82448322025e084cadc56048b4e",
+                "shasum": ""
+            },
+            "require": {
+                "php": ">=7.1"
+            },
+            "suggest": {
+                "ext-ctype": "For best performance"
+            },
+            "default-branch": true,
+            "type": "library",
+            "extra": {
+                "branch-alias": {
+                    "dev-main": "1.22-dev"
+                },
+                "thanks": {
+                    "name": "symfony/polyfill",
+                    "url": "https://github.com/symfony/polyfill"
+                }
+            },
+            "autoload": {
+                "psr-4": {
+                    "Symfony\\Polyfill\\Ctype\\": ""
+                },
+                "files": [
+                    "bootstrap.php"
+                ]
+            },
+            "notification-url": "https://packagist.org/downloads/",
+            "license": [
+                "MIT"
+            ],
+            "authors": [
+                {
+                    "name": "Gert de Pagter",
+                    "email": "BackEndTea@gmail.com"
+                },
+                {
+                    "name": "Symfony Community",
+                    "homepage": "https://symfony.com/contributors"
+                }
+            ],
+            "description": "Symfony polyfill for ctype functions",
+            "homepage": "https://symfony.com",
+            "keywords": [
+                "compatibility",
+                "ctype",
+                "polyfill",
+                "portable"
+            ],
+            "support": {
+                "source": "https://github.com/symfony/polyfill-ctype/tree/v1.22.1"
+            },
+            "funding": [
+                {
+                    "url": "https://symfony.com/sponsor",
+                    "type": "custom"
+                },
+                {
+                    "url": "https://github.com/fabpot",
+                    "type": "github"
+                },
+                {
+                    "url": "https://tidelift.com/funding/github/packagist/symfony/symfony",
+                    "type": "tidelift"
+                }
+            ],
+            "time": "2021-01-07T16:49:33+00:00"
+        },
         {
             "name": "symfony/polyfill-intl-grapheme",
             "version": "dev-main",
@@ -5603,6 +5683,55 @@
             ],
             "time": "2021-02-17T15:27:35+00:00"
         },
+        {
+            "name": "textalk/websocket",
+            "version": "1.5.2",
+            "source": {
+                "type": "git",
+                "url": "https://github.com/Textalk/websocket-php.git",
+                "reference": "b93249453806a2dd46495de46d76fcbcb0d8dee8"
+            },
+            "dist": {
+                "type": "zip",
+                "url": "https://api.github.com/repos/Textalk/websocket-php/zipball/b93249453806a2dd46495de46d76fcbcb0d8dee8",
+                "reference": "b93249453806a2dd46495de46d76fcbcb0d8dee8",
+                "shasum": ""
+            },
+            "require": {
+                "php": "^7.2 | ^8.0",
+                "psr/log": "^1.0"
+            },
+            "require-dev": {
+                "php-coveralls/php-coveralls": "^2.0",
+                "phpunit/phpunit": "^8.0|^9.0",
+                "squizlabs/php_codesniffer": "^3.5"
+            },
+            "type": "library",
+            "autoload": {
+                "psr-4": {
+                    "WebSocket\\": "lib"
+                }
+            },
+            "notification-url": "https://packagist.org/downloads/",
+            "license": [
+                "ISC"
+            ],
+            "authors": [
+                {
+                    "name": "Fredrik Liljegren"
+                },
+                {
+                    "name": "Sören Jensen",
+                    "email": "soren@abicart.se"
+                }
+            ],
+            "description": "WebSocket client and server",
+            "support": {
+                "issues": "https://github.com/Textalk/websocket-php/issues",
+                "source": "https://github.com/Textalk/websocket-php/tree/1.5.2"
+            },
+            "time": "2021-02-12T15:39:23+00:00"
+        },
         {
             "name": "theseer/tokenizer",
             "version": "1.2.0",
@@ -5837,6 +5966,65 @@
             },
             "time": "2020-11-02T05:54:12+00:00"
         },
+        {
+            "name": "webmozart/assert",
+            "version": "dev-master",
+            "source": {
+                "type": "git",
+                "url": "https://github.com/webmozarts/assert.git",
+                "reference": "4631e2c7d2d7132adac9fd84d4c1a98c10a6e049"
+            },
+            "dist": {
+                "type": "zip",
+                "url": "https://api.github.com/repos/webmozarts/assert/zipball/4631e2c7d2d7132adac9fd84d4c1a98c10a6e049",
+                "reference": "4631e2c7d2d7132adac9fd84d4c1a98c10a6e049",
+                "shasum": ""
+            },
+            "require": {
+                "php": "^7.2 || ^8.0",
+                "symfony/polyfill-ctype": "^1.8"
+            },
+            "conflict": {
+                "phpstan/phpstan": "<0.12.20",
+                "vimeo/psalm": "<3.9.1"
+            },
+            "require-dev": {
+                "phpunit/phpunit": "^8.5.13"
+            },
+            "default-branch": true,
+            "type": "library",
+            "extra": {
+                "branch-alias": {
+                    "dev-master": "1.10-dev"
+                }
+            },
+            "autoload": {
+                "psr-4": {
+                    "Webmozart\\Assert\\": "src/"
+                }
+            },
+            "notification-url": "https://packagist.org/downloads/",
+            "license": [
+                "MIT"
+            ],
+            "authors": [
+                {
+                    "name": "Bernhard Schussek",
+                    "email": "bschussek@gmail.com"
+                }
+            ],
+            "description": "Assertions to validate method input/output with nice error messages.",
+            "keywords": [
+                "assert",
+                "check",
+                "validate"
+            ],
+            "support": {
+                "issues": "https://github.com/webmozarts/assert/issues",
+                "source": "https://github.com/webmozarts/assert/tree/master"
+            },
+            "time": "2021-02-28T20:01:57+00:00"
+        },
         {
             "name": "webmozart/path-util",
             "version": "dev-master",

src/Appwrite/Realtime/Realtime.php

@@ -87,6 +87,14 @@ class Realtime
     /**
      * Identifies the receivers of all subscriptions, based on the permissions and event.
      * 
+     * Example of performance with an event with user:XXX permissions and with X users spread across 10 different channels:
+     *  - 0.014 ms (±6.88%) | 10 Connections / 100 Subscriptions 
+     *  - 0.070 ms (±3.71%) | 100 Connections / 1,000 Subscriptions 
+     *  - 0.846 ms (±2.74%) | 1,000 Connections / 10,000 Subscriptions
+     *  - 10.866 ms (±1.01%) | 10,000 Connections / 100,000 Subscriptions
+     *  - 110.201 ms (±2.32%) | 100,000 Connections / 1,000,000 Subscriptions
+     *  - 1,121.328 ms (±0.84%) | 1,000,000 Connections / 10,000,000 Subscriptions 
+     * 
      * @param array $event
      * @param array $connections
      * @param array $subscriptions

tests/e2e/Services/Realtime/RealtimeBase.php

@@ -0,0 +1,593 @@
+<?php
+
+namespace Tests\E2E\Services\Realtime;
+
+use CURLFile;
+use Tests\E2E\Client;
+use WebSocket\Client as WebSocketClient;
+use WebSocket\ConnectionException;
+
+trait RealtimeBase
+{
+
+    private function getWebsocket($channels = [], $headers = [])
+    {
+        $headers = array_merge([
+            'Origin' => 'appwrite.test'
+        ], $headers);
+
+        $query = [
+            'project' => $this->getProject()['$id'],
+            'channels' => $channels
+        ];
+        return new WebSocketClient('ws://appwrite-traefik/v1/realtime?' . http_build_query($query), [
+            'headers' => $headers,
+            'timeout' => 5,
+        ]);
+    }
+
+    public function testConnection()
+    {
+        /**
+         * Test for SUCCESS
+         */
+        $client = $this->getWebsocket(['documents']);
+        $this->assertEquals(json_encode([
+            'documents' => 0
+        ]), $client->receive());
+        $client->close();
+
+        /**
+         * Test for FAILURE
+         */
+        $client = $this->getWebsocket(['documents'], ['origin' => 'http://appwrite.unknown']);
+        $this->assertEquals('Invalid Origin. Register your new client (appwrite.unknown) as a new Web platform on your project console dashboard', $client->receive());
+        $this->expectException(ConnectionException::class); // Check if server disconnnected client
+        $client->close();
+
+        $client = $this->getWebsocket();
+        $this->assertEquals('Missing channels', $client->receive());
+        $this->expectException(ConnectionException::class); // Check if server disconnnected client
+        $client->close();
+
+        $client = new WebSocketClient('ws://appwrite-traefik/v1/realtime', [
+            'headers' => [
+                'Origin' => 'appwrite.test'
+            ]
+        ]);
+        $this->assertEquals('Missing or unknown project ID', $client->receive());
+        $this->expectException(ConnectionException::class); // Check if server disconnnected client
+        $client->close();
+
+        $client = new WebSocketClient('ws://appwrite-traefik/v1/realtime?project=123', [
+            'headers' => [
+                'Origin' => 'appwrite.test'
+            ]
+        ]);
+        $this->assertEquals('Missing or unknown project ID', $client->receive());
+        $this->expectException(ConnectionException::class); // Check if server disconnnected client
+        $client->close();
+    }
+
+    public function testChannelParsing()
+    {
+        $user = $this->getUser();
+        $userId = $user['$id'] ?? '';
+        $session = $user['session'] ?? '';
+        $headers =  [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_'.$this->getProject()['$id'].'=' . $session
+        ];
+
+        $client = $this->getWebsocket(['documents']);
+        $response = json_decode($client->receive(), true);
+        $this->assertCount(1, $response);
+        $this->assertArrayHasKey('documents', $response);
+        $client->close();
+
+        $client = $this->getWebsocket(['account'], $headers);
+        $response = json_decode($client->receive(), true);
+        $this->assertCount(1, $response);
+        $this->assertArrayHasKey('account.' . $userId, $response);
+        $client->close();
+
+        $client = $this->getWebsocket(['account', 'documents', 'account.123'], $headers);
+        $response = json_decode($client->receive(), true);
+        $this->assertCount(2, $response);
+        $this->assertArrayHasKey('documents', $response);
+        $this->assertArrayHasKey('account.' . $userId, $response);
+        $client->close();
+
+        $client = $this->getWebsocket([
+            'account',
+            'files',
+            'files.1',
+            'collections',
+            'collections.1',
+            'collections.1.documents',
+            'collections.2',
+            'collections.2.documents',
+            'documents',
+            'documents.1',
+            'documents.2',
+        ], $headers);
+        $response = json_decode($client->receive(), true);
+
+        $this->assertCount(11, $response);
+        $this->assertArrayHasKey('account.' . $userId, $response);
+        $this->assertArrayHasKey('files', $response);
+        $this->assertArrayHasKey('files.1', $response);
+        $this->assertArrayHasKey('collections', $response);
+        $this->assertArrayHasKey('collections.1', $response);
+        $this->assertArrayHasKey('collections.1.documents', $response);
+        $this->assertArrayHasKey('collections.2', $response);
+        $this->assertArrayHasKey('collections.2.documents', $response);
+        $this->assertArrayHasKey('documents', $response);
+        $this->assertArrayHasKey('documents.1', $response);
+        $this->assertArrayHasKey('documents.2', $response);
+        $client->close();
+    }
+
+    public function testChannelAccount()
+    {
+        $user = $this->getUser();
+        $userId = $user['$id'] ?? '';
+        $session = $user['session'] ?? '';
+        $projectId = $this->getProject()['$id'];
+
+        $client = $this->getWebsocket(['account'], [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_'.$projectId.'=' . $session
+        ]);
+        $response = json_decode($client->receive(), true);
+        $this->assertCount(1, $response);
+        $this->assertArrayHasKey('account.' . $userId, $response);
+
+        /**
+         * Test Account Name Event
+         */
+        $name = "Torsten Dittmann";
+
+        $this->client->call(Client::METHOD_PATCH, '/account/name', array_merge([
+            'origin' => 'http://localhost',
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+            'cookie' => 'a_session_' . $projectId . '=' . $session,
+        ]), [
+            'name' => $name
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertCount(1, $response['channels']);
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertEquals('account.' . $userId, $response['channels'][0]);
+        $this->assertEquals('account.update.name', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        $this->assertEquals($name, $response['payload']['name']);
+
+
+        /**
+         * Test Account Password Event
+         */
+        $this->client->call(Client::METHOD_PATCH, '/account/password', array_merge([
+            'origin' => 'http://localhost',
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+            'cookie' => 'a_session_'.$projectId.'=' . $session,
+        ]), [
+            'password' => 'new-password',
+            'oldPassword' => 'password',
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertCount(1, $response['channels']);
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertEquals('account.' . $userId, $response['channels'][0]);
+        $this->assertEquals('account.update.password', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        $this->assertEquals($name, $response['payload']['name']);
+
+        /**
+         * Test Account Email Update
+         */
+        $this->client->call(Client::METHOD_PATCH, '/account/email', array_merge([
+            'origin' => 'http://localhost',
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+            'cookie' => 'a_session_'.$projectId.'=' . $session,
+        ]), [
+            'email' => 'torsten@appwrite.io',
+            'password' => 'new-password',
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertCount(1, $response['channels']);
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertEquals('account.' . $userId, $response['channels'][0]);
+        $this->assertEquals('account.update.email', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        $this->assertEquals('torsten@appwrite.io', $response['payload']['email']);
+
+        /**
+         * Test Account Verification Create
+         */
+        $this->client->call(Client::METHOD_POST, '/account/verification', array_merge([
+            'origin' => 'http://localhost',
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+            'cookie' => 'a_session_'.$projectId.'=' . $session,
+        ]), [
+            'url' => 'http://localhost/verification',
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertCount(1, $response['channels']);
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertEquals('account.' . $userId, $response['channels'][0]);
+        $this->assertEquals('account.verification.create', $response['event']);
+
+        $lastEmail = $this->getLastEmail();
+        $verification = substr($lastEmail['text'], strpos($lastEmail['text'], '&secret=', 0) + 8, 256);
+
+        /**
+         * Test Account Verification Complete
+         */
+        $response = $this->client->call(Client::METHOD_PUT, '/account/verification', array_merge([
+            'origin' => 'http://localhost',
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+            'cookie' => 'a_session_'.$projectId.'=' . $session,
+        ]), [
+            'userId' => $userId,
+            'secret' => $verification,
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertCount(1, $response['channels']);
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertEquals('account.' . $userId, $response['channels'][0]);
+        $this->assertEquals('account.verification.update', $response['event']);
+
+        /**
+         * Test Acoount Prefs Update
+         */
+        $this->client->call(Client::METHOD_PATCH, '/account/prefs', array_merge([
+            'origin' => 'http://localhost',
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+            'cookie' => 'a_session_'.$projectId.'=' . $session,
+        ]), [
+            'prefs' => [
+                'prefKey1' => 'prefValue1',
+                'prefKey2' => 'prefValue2',
+            ]
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertCount(1, $response['channels']);
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertEquals('account.' . $userId, $response['channels'][0]);
+        $this->assertEquals('account.update.prefs', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        /**
+         * Test Account Session Create
+         */
+        $response = $this->client->call(Client::METHOD_POST, '/account/sessions', array_merge([
+            'origin' => 'http://localhost',
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ]), [
+            'email' => 'torsten@appwrite.io',
+            'password' => 'new-password',
+        ]);
+
+        $sessionNew = $this->client->parseCookie((string)$response['headers']['set-cookie'])['a_session_'.$projectId];
+        $sessionNewId = $response['body']['$id'];
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertCount(1, $response['channels']);
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertEquals('account.' . $userId, $response['channels'][0]);
+        $this->assertEquals('account.sessions.create', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        /**
+         * Test Account Session Delete
+         */
+        $this->client->call(Client::METHOD_DELETE, '/account/sessions/'.$sessionNewId, array_merge([
+            'origin' => 'http://localhost',
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+            'cookie' => 'a_session_'.$projectId.'=' . $sessionNew,
+        ]));
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertCount(1, $response['channels']);
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertEquals('account.' . $userId, $response['channels'][0]);
+        $this->assertEquals('account.sessions.delete', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        /**
+         * Test Account Create Recovery
+         */
+        $this->client->call(Client::METHOD_POST, '/account/recovery', array_merge([
+            'origin' => 'http://localhost',
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ]), [
+            'email' => 'torsten@appwrite.io',
+            'url' => 'http://localhost/recovery',
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $lastEmail = $this->getLastEmail();
+        $recovery = substr($lastEmail['text'], strpos($lastEmail['text'], '&secret=', 0) + 8, 256);
+
+        $this->assertCount(1, $response['channels']);
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertEquals('account.' . $userId, $response['channels'][0]);
+        $this->assertEquals('account.recovery.create', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        $response = $this->client->call(Client::METHOD_PUT, '/account/recovery', array_merge([
+            'origin' => 'http://localhost',
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ]), [
+            'userId' => $userId,
+            'secret' => $recovery,
+            'password' => 'test-recovery',
+            'passwordAgain' => 'test-recovery',
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertCount(1, $response['channels']);
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertEquals('account.' . $userId, $response['channels'][0]);
+        $this->assertEquals('account.recovery.update', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        $client->close();
+    }
+
+    public function testChannelDatabase()
+    {
+        $user = $this->getUser();
+        $session = $user['session'] ?? '';
+        $projectId = $this->getProject()['$id'];
+
+        $client = $this->getWebsocket(['documents', 'collections'], [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_'.$projectId.'=' . $session
+        ]);
+        $response = json_decode($client->receive(), true);
+        $this->assertCount(2, $response);
+        $this->assertArrayHasKey('documents', $response);
+        $this->assertArrayHasKey('collections', $response);
+
+        /**
+         * Test Collection Create
+         */
+        $actors = $this->client->call(Client::METHOD_POST, '/database/collections', array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey']
+        ]), [
+            'name' => 'Actors',
+            'read' => ['*'],
+            'write' => ['*'],
+            'rules' => [
+                [
+                    'label' => 'First Name',
+                    'key' => 'firstName',
+                    'type' => 'text',
+                    'default' => '',
+                    'required' => true,
+                    'array' => false
+                ],
+                [
+                    'label' => 'Last Name',
+                    'key' => 'lastName',
+                    'type' => 'text',
+                    'default' => '',
+                    'required' => true,
+                    'array' => false
+                ],
+            ],
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertCount(2, $response['channels']);
+        $this->assertContains('collections', $response['channels']);
+        $this->assertContains('collections.' . $actors['body']['$id'], $response['channels']);
+        $this->assertEquals('database.collections.create', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        $data = ['actorsId' => $actors['body']['$id']];
+
+        /**
+         * Test Document Create
+         */
+        $document = $this->client->call(Client::METHOD_POST, '/database/collections/' . $data['actorsId'] . '/documents', array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+        ], $this->getHeaders()), [
+            'data' => [
+                'firstName' => 'Chris',
+                'lastName' => 'Evans',
+            ],
+            'read' => ['*'],
+            'write' => ['*'],
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertCount(3, $response['channels']);
+        $this->assertContains('documents', $response['channels']);
+        $this->assertContains('documents.' . $document['body']['$id'], $response['channels']);
+        $this->assertContains('collections.' . $actors['body']['$id'] . '.documents', $response['channels']);
+        $this->assertEquals('database.documents.create', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        $data['documentId'] = $document['body']['$id'];
+
+        /**
+         * Test Document Update
+         */
+        $document = $this->client->call(Client::METHOD_PATCH, '/database/collections/' . $data['actorsId'] . '/documents/' . $data['documentId'], array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+        ], $this->getHeaders()), [
+            'data' => [
+                'firstName' => 'Chris1',
+                'lastName' => 'Evans2',
+            ],
+            'read' => ['*'],
+            'write' => ['*'],
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertCount(3, $response['channels']);
+        $this->assertContains('documents', $response['channels']);
+        $this->assertContains('documents.' . $data['documentId'], $response['channels']);
+        $this->assertContains('collections.' . $data['actorsId'] . '.documents', $response['channels']);
+        $this->assertEquals('database.documents.update', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        $this->assertEquals($response['payload']['firstName'], 'Chris1');
+        $this->assertEquals($response['payload']['lastName'], 'Evans2');
+
+        /**
+         * Test Document Delete
+         */
+        $document = $this->client->call(Client::METHOD_POST, '/database/collections/' . $data['actorsId'] . '/documents', array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+        ], $this->getHeaders()), [
+            'data' => [
+                'firstName' => 'Bradly',
+                'lastName' => 'Cooper',
+
+            ],
+            'read' => ['*'],
+            'write' => ['*'],
+        ]);
+
+        $client->receive();
+
+        $this->client->call(Client::METHOD_DELETE, '/database/collections/' . $data['actorsId'] . '/documents/' . $document['body']['$id'], array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+        ], $this->getHeaders()));
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertCount(3, $response['channels']);
+        $this->assertContains('documents', $response['channels']);
+        $this->assertContains('documents.' . $document['body']['$id'], $response['channels']);
+        $this->assertContains('collections.' . $data['actorsId'] . '.documents', $response['channels']);
+        $this->assertEquals('database.documents.delete', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        $client->close();
+    }
+
+    public function testChannelFiles()
+    {
+        $user = $this->getUser();
+        $session = $user['session'] ?? '';
+        $projectId = $this->getProject()['$id'];
+
+        $client = $this->getWebsocket(['files'], [
+            'origin' => 'http://localhost',
+            'cookie' => 'a_session_'.$projectId.'=' . $session
+        ]);
+        $response = json_decode($client->receive(), true);
+        $this->assertCount(1, $response);
+        $this->assertArrayHasKey('files', $response);
+
+        /**
+         * Test File Create
+         */
+        $file = $this->client->call(Client::METHOD_POST, '/storage/files', array_merge([
+            'content-type' => 'multipart/form-data',
+            'x-appwrite-project' => $this->getProject()['$id'],
+        ], $this->getHeaders()), [
+            'file' => new CURLFile(realpath(__DIR__ . '/../../../resources/logo.png'), 'image/png', 'logo.png'),
+            'read' => ['*'],
+            'write' => ['*'],
+            'folderId' => 'xyz',
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertCount(2, $response['channels']);
+        $this->assertContains('files', $response['channels']);
+        $this->assertContains('files.' . $file['body']['$id'], $response['channels']);
+        $this->assertEquals('storage.files.create', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        $data = ['fileId' => $file['body']['$id']];
+
+        /**
+         * Test File Update
+         */
+        $this->client->call(Client::METHOD_PUT, '/storage/files/' . $data['fileId'], array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+        ], $this->getHeaders()), [
+            'read' => ['*'],
+            'write' => ['*'],
+        ]);
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertCount(2, $response['channels']);
+        $this->assertContains('files', $response['channels']);
+        $this->assertContains('files.' . $file['body']['$id'], $response['channels']);
+        $this->assertEquals('storage.files.update', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        /**
+         * Test File Delete
+         */
+        $this->client->call(Client::METHOD_DELETE, '/storage/files/' . $data['fileId'], array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+        ], $this->getHeaders()));
+
+        $response = json_decode($client->receive(), true);
+
+        $this->assertArrayHasKey('timestamp', $response);
+        $this->assertCount(2, $response['channels']);
+        $this->assertContains('files', $response['channels']);
+        $this->assertContains('files.' . $file['body']['$id'], $response['channels']);
+        $this->assertEquals('storage.files.delete', $response['event']);
+        $this->assertNotEmpty($response['payload']);
+
+        $client->close();
+    }
+}

tests/e2e/Services/Realtime/RealtimeCustomClientTest.php

@@ -0,0 +1,15 @@
+<?php
+
+namespace Tests\E2E\Services\Realtime;
+
+use Tests\E2E\Scopes\Scope;
+use Tests\E2E\Scopes\ProjectCustom;
+use Tests\E2E\Scopes\SideClient;
+
+
+class RealtimeCustomClientTest extends Scope
+{
+    use RealtimeBase;
+    use ProjectCustom;
+    use SideClient;
+}