Merge pull request #2786 from appwrite/feat-update-realtime-events

storage realtime event refactor

app/controllers/api/storage.php

@@ -358,15 +358,17 @@ App::post('/v1/storage/buckets/:bucketId/files')
     ->inject('user')
     ->inject('audits')
     ->inject('usage')
+    ->inject('events')
     ->inject('mode')
     ->inject('deviceFiles')
     ->inject('deviceLocal')
-    ->action(function ($bucketId, $fileId, $file, $read, $write, $request, $response, $dbForProject, $user, $audits, $usage, $mode, $deviceFiles, $deviceLocal) {
+    ->action(function ($bucketId, $fileId, $file, $read, $write, $request, $response, $dbForProject, $user, $audits, $usage, $events, $mode, $deviceFiles, $deviceLocal) {
         /** @var Utopia\Swoole\Request $request */
         /** @var Appwrite\Utopia\Response $response */
         /** @var Utopia\Database\Database $dbForProject */
         /** @var Utopia\Database\Document $user */
         /** @var Appwrite\Event\Event $audits */
+        /** @var Appwrite\Event\Event $events */
         /** @var Appwrite\Stats\Stats $usage */
         /** @var Utopia\Storage\Device $deviceFiles */
         /** @var Utopia\Storage\Device $deviceLocal */
@@ -673,6 +675,8 @@ App::post('/v1/storage/buckets/:bucketId/files')
             }
         }
 
+        $events->setParam('bucket', $bucket->getArrayCopy());
+
         $metadata = null; // was causing leaks as it was passed by reference
 
         $response->setStatusCode(Response::STATUS_CODE_CREATED);
@@ -1339,12 +1343,14 @@ App::put('/v1/storage/buckets/:bucketId/files/:fileId')
     ->inject('audits')
     ->inject('usage')
     ->inject('mode')
-    ->action(function ($bucketId, $fileId, $read, $write, $response, $dbForProject, $user, $audits, $usage, $mode) {
+    ->inject('events')
+    ->action(function ($bucketId, $fileId, $read, $write, $response, $dbForProject, $user, $audits, $usage, $mode, $events) {
         /** @var Appwrite\Utopia\Response $response */
         /** @var Utopia\Database\Database $dbForProject */
         /** @var Utopia\Database\Document $user */
         /** @var Appwrite\Event\Event $audits */
         /** @var Appwrite\Stats\Stats $usage */
+        /** @var Appwirte\Event\Event $events */
         /** @var string $mode */
 
         $bucket = $dbForProject->getDocument('buckets', $bucketId);
@@ -1406,6 +1412,8 @@ App::put('/v1/storage/buckets/:bucketId/files/:fileId')
             );
         }
 
+        $events->setParam('bucket', $bucket->getArrayCopy());
+
         $audits
             ->setParam('event', 'storage.files.update')
             ->setParam('resource', 'file/' . $file->getId())
@@ -1521,6 +1529,7 @@ App::delete('/v1/storage/buckets/:bucketId/files/:fileId')
 
         $events
             ->setParam('eventData', $response->output($file, Response::MODEL_FILE))
+            ->setParam('bucket', $bucket->getArrayCopy())
         ;
 
         $response->noContent();

app/controllers/shared/api.php

@@ -217,12 +217,14 @@ App::shutdown(function ($utopia, $request, $response, $project, $events, $audits
         if ($project->getId() !== 'console') {
             $payload = new Document($response->getPayload());
             $collection = new Document($events->getParam('collection') ?? []);
+            $bucket = new Document($events->getParam('bucket') ?? []);
 
             $target = Realtime::fromPayload(
                 event: $events->getParam('event'), 
                 payload: $payload, 
                 project: $project, 
-                collection: $collection
+                collection: $collection,
+                bucket: $bucket,
             );
 
             Realtime::send(

src/Appwrite/Messaging/Adapter/Realtime.php

@@ -240,7 +240,7 @@ class Realtime extends Adapter
      * @param Document|null $project 
      * @return array 
      */
-    public static function fromPayload(string $event, Document $payload, Document $project = null, Document $collection = null): array
+    public static function fromPayload(string $event, Document $payload, Document $project = null, Document $collection = null, Document $bucket = null): array
     {
         $channels = [];
         $roles = [];
@@ -285,7 +285,7 @@ class Realtime extends Adapter
                 break;
             case strpos($event, 'database.documents.') === 0:
                 if ($collection->isEmpty()) {
-                    throw new \Exception('Collection need to be passed to to Realtime for Document events in the Database.');
+                    throw new \Exception('Collection needs to be passed to Realtime for Document events in the Database.');
                 }
 
                 $channels[] = 'documents';
@@ -294,17 +294,14 @@ class Realtime extends Adapter
 
                 $roles = ($collection->getAttribute('permission') === 'collection') ? $collection->getRead() : $payload->getRead();
 
-                break;
-            case strpos($event, 'storage.buckets.') === 0:
-                $channels[] = 'buckets';
-                $channels[] = 'buckets.' . $payload->getId();
-                $roles = $payload->getRead();
-
                 break;
             case strpos($event, 'storage.files') === 0:
+                if($bucket->isEmpty()) {
+                    throw new \Exception('Bucket needs to be pased to Realtime for File events in the Storage.');
+                }
                 $channels[] = 'files';
                 $channels[] = 'buckets.' . $payload->getAttribute('bucketId') . '.files';
-                $channels[] = 'files.' . $payload->getId();
+                $channels[] = 'buckets.' . $payload->getAttribute('bucketId') . '.files.' . $payload->getId();
                 $roles = $payload->getRead();
 
                 break;

tests/e2e/Services/Realtime/RealtimeCustomClientTest.php

@@ -851,7 +851,7 @@ class RealtimeCustomClientTest extends Scope
         $session = $user['session'] ?? '';
         $projectId = $this->getProject()['$id'];
 
-        $client = $this->getWebsocket(['files', 'buckets'], [
+        $client = $this->getWebsocket(['files'], [
             'origin' => 'http://localhost',
             'cookie' => 'a_session_'.$projectId.'=' . $session
         ]);
@@ -861,9 +861,8 @@ class RealtimeCustomClientTest extends Scope
         $this->assertArrayHasKey('data', $response);
         $this->assertEquals('connected', $response['type']);
         $this->assertNotEmpty($response['data']);
-        $this->assertCount(2, $response['data']['channels']);
+        $this->assertCount(1, $response['data']['channels']);
         $this->assertContains('files', $response['data']['channels']);
-        $this->assertContains('buckets', $response['data']['channels']);
         $this->assertNotEmpty($response['data']['user']);
         $this->assertEquals($user['$id'], $response['data']['user']['$id']);
 
@@ -881,18 +880,7 @@ class RealtimeCustomClientTest extends Scope
             'write' => ['role:all'],
             'permission' => 'bucket'
         ]);
-        $response = json_decode($client->receive(), true);
 
-        $this->assertArrayHasKey('type', $response);
-        $this->assertArrayHasKey('data', $response);
-        $this->assertEquals('event', $response['type']);
-        $this->assertNotEmpty($response['data']);
-        $this->assertArrayHasKey('timestamp', $response['data']);
-        $this->assertCount(2, $response['data']['channels']);
-        $this->assertContains('buckets', $response['data']['channels']);
-        $this->assertContains('buckets.' . $bucket1['body']['$id'], $response['data']['channels']);
-        $this->assertEquals('storage.buckets.create', $response['data']['event']);
-        $this->assertNotEmpty($response['data']['payload']);
         $data = ['bucketId' => $bucket1['body']['$id']];
         /**
          * Test File Create
@@ -905,7 +893,6 @@ class RealtimeCustomClientTest extends Scope
             'file' => new CURLFile(realpath(__DIR__ . '/../../../resources/logo.png'), 'image/png', 'logo.png'),
             'read' => ['role:all'],
             'write' => ['role:all'],
-            'folderId' => 'xyz',
         ]);
 
         $response = json_decode($client->receive(), true);
@@ -917,7 +904,7 @@ class RealtimeCustomClientTest extends Scope
         $this->assertArrayHasKey('timestamp', $response['data']);
         $this->assertCount(3, $response['data']['channels']);
         $this->assertContains('files', $response['data']['channels']);
-        $this->assertContains('files.' . $file['body']['$id'], $response['data']['channels']);
+        $this->assertContains('buckets.' . $data['bucketId'] . '.files.' . $file['body']['$id'], $response['data']['channels']);
         $this->assertContains('buckets.' . $data['bucketId'] . '.files', $response['data']['channels']);
         $this->assertEquals('storage.files.create', $response['data']['event']);
         $this->assertNotEmpty($response['data']['payload']);
@@ -944,7 +931,7 @@ class RealtimeCustomClientTest extends Scope
         $this->assertArrayHasKey('timestamp', $response['data']);
         $this->assertCount(3, $response['data']['channels']);
         $this->assertContains('files', $response['data']['channels']);
-        $this->assertContains('files.' . $file['body']['$id'], $response['data']['channels']);
+        $this->assertContains('buckets.' . $data['bucketId'] . '.files.' . $file['body']['$id'], $response['data']['channels']);
         $this->assertContains('buckets.' . $data['bucketId'] . '.files', $response['data']['channels']);
         $this->assertEquals('storage.files.update', $response['data']['event']);
         $this->assertNotEmpty($response['data']['payload']);
@@ -966,7 +953,7 @@ class RealtimeCustomClientTest extends Scope
         $this->assertArrayHasKey('timestamp', $response['data']);
         $this->assertCount(3, $response['data']['channels']);
         $this->assertContains('files', $response['data']['channels']);
-        $this->assertContains('files.' . $file['body']['$id'], $response['data']['channels']);
+        $this->assertContains('buckets.' . $data['bucketId'] . '.files.' . $file['body']['$id'], $response['data']['channels']);
         $this->assertContains('buckets.' . $data['bucketId'] . '.files', $response['data']['channels']);
         $this->assertEquals('storage.files.delete', $response['data']['event']);
         $this->assertNotEmpty($response['data']['payload']);