added presence api update

mock-server/app/http.php

@@ -6,9 +6,7 @@
     require_once __DIR__ . '/../vendor/autoload.php';
 }
 
-use Swoole\Constant;
 use Utopia\App;
-use Utopia\Database\Document;
 use Utopia\Database\Helpers\ID;
 use Utopia\MockServer\Utopia\Exception;
 use Utopia\MockServer\Utopia\File;
@@ -24,10 +22,11 @@
 use Utopia\Validator\Host;
 use Utopia\Validator\Nullable;
 use Utopia\Validator\WhiteList;
-use Swoole\Process;
-use Swoole\Http\Server;
 use Utopia\MockServer\Utopia\Model\Player;
 use Utopia\MockServer\Utopia\Validator\Player as PlayerValidator;
+use Utopia\MockServer\Utopia\Realtime\Protocol as RealtimeProtocol;
+use Utopia\WebSocket\Adapter\Swoole;
+use Utopia\WebSocket\Server as WebSocketServer;
 
 const APP_AUTH_TYPE_SESSION = 'Session';
 const APP_AUTH_TYPE_JWT = 'JWT';
@@ -39,23 +38,24 @@
 const APP_PLATFORM_CONSOLE = 'console';
 const APP_STORAGE_CACHE = '/storage/cache';
 
-$http = new Server(
-    host: '0.0.0.0',
-    port: App::getEnv('PORT', 80),
-    mode: SWOOLE_PROCESS
-);
 $payloadSize = 6 * (1024 * 1024); // 6MB
 $workerNumber = swoole_cpu_num() * intval(App::getEnv('_APP_WORKER_PER_CORE', 6));
 
-$http->set([
-    'worker_num' => $workerNumber,
+$adapter = new Swoole(host: '0.0.0.0', port: (int) App::getEnv('PORT', 80));
+$adapter
+    ->setPackageMaxLength($payloadSize)
+    ->setWorkerNumber($workerNumber);
+
+// Settings the adapter doesn't expose directly.
+$adapter->getNative()->set([
     'open_http2_protocol' => true,
     'http_compression' => true,
     'http_compression_level' => 6,
-    'package_max_length' => $payloadSize,
     'buffer_output_size' => $payloadSize,
 ]);
 
+$server = new WebSocketServer($adapter);
+
 // Version Route for CLI
 App::get('/v1/health/version')
     ->desc('Get version')
@@ -862,21 +862,23 @@ function ($utopia, $error, $request, $response) {
         ['utopia', 'error', 'request', 'response']
     );
 
-$http->on(Constant::EVENT_START, function (Server $http) use ($payloadSize) {
+/**
+ * Realtime WebSocket mock at /v1/realtime?project=<id>.
+ *
+ * Single Protocol instance is shared across worker invocations within the same
+ * worker process; per-connection state lives inside it keyed by Swoole fd.
+ */
+$realtimeProtocol = new RealtimeProtocol();
+
+$server->error(function (\Throwable $error, string $action) {
+    Console::error("[ws:{$action}] " . $error->getMessage());
+});
+
+$server->onStart(function () use ($payloadSize) {
     Console::success('Server started successfully (max payload is ' . number_format($payloadSize) . ' bytes)');
-    Console::info("Master pid {$http->master_pid}, manager pid {$http->manager_pid}");
-
-    // Listen for ctrl + c
-    Process::signal(
-        2,
-        function () use ($http) {
-            Console::log('Stop by Ctrl+C');
-            $http->shutdown();
-        }
-    );
 });
 
-$http->on(Constant::EVENT_REQUEST, function (SwooleRequest $swooleRequest, SwooleResponse $swooleResponse) {
+$server->onRequest(function (SwooleRequest $swooleRequest, SwooleResponse $swooleResponse) {
     $request = new Request($swooleRequest);
     $response = new UtopiaSwooleResponse($swooleResponse);
 
@@ -885,4 +887,29 @@ function () use ($http) {
     $app->run($request, $response);
 });
 
-$http->start();
+$server->onOpen(function (int $fd, SwooleRequest $swooleRequest) use ($server, $realtimeProtocol) {
+    $path = (string) ($swooleRequest->server['request_uri'] ?? '');
+    if ($path !== '/v1/realtime') {
+        // Reject upgrades on any other path with a policy-violation close.
+        $server->close($fd, 1008);
+        return;
+    }
+
+    $project = (string) ($swooleRequest->get['project'] ?? '');
+    if ($project === '') {
+        $server->close($fd, 1008);
+        return;
+    }
+
+    $realtimeProtocol->open($server, $fd, $project);
+});
+
+$server->onMessage(function (int $fd, string $data) use ($server, $realtimeProtocol) {
+    $realtimeProtocol->message($server, $fd, $data);
+});
+
+$server->onClose(function (int $fd) use ($realtimeProtocol) {
+    $realtimeProtocol->close($fd);
+});
+
+$server->start();

mock-server/composer.json

@@ -10,7 +10,9 @@
         "utopia-php/framework": "0.33.*",
         "utopia-php/database": "0.48.*",
         "utopia-php/cli": "0.16.*",
-        "utopia-php/swoole": "0.8.*"
+        "utopia-php/swoole": "0.8.*",
+        "utopia-php/websocket": "^1.0",
+        "utopia-php/query": "^0.3.1"
     },
     "require-dev": {
         "swoole/ide-helper": "5.1.2"

mock-server/src/Utopia/Realtime/Connection.php

@@ -0,0 +1,59 @@
+<?php
+
+namespace Utopia\MockServer\Utopia\Realtime;
+
+/**
+ * Per-connection state for a single WebSocket client.
+ *
+ * One instance lives per Swoole `fd` for as long as the socket is open.
+ */
+class Connection
+{
+    public int $fd;
+    public string $project = '';
+    public ?array $user = null;
+
+    /**
+     * Active subscriptions keyed by client-supplied subscriptionId.
+     * Each value is `['channels' => string[], 'queries' => string[]]`.
+     *
+     * @var array<string, array{channels: string[], queries: string[]}>
+     */
+    public array $subscriptions = [];
+
+    public function __construct(int $fd, string $project = '')
+    {
+        $this->fd = $fd;
+        $this->project = $project;
+    }
+
+    public function subscribe(string $subscriptionId, array $channels, array $queries): void
+    {
+        $this->subscriptions[$subscriptionId] = [
+            'channels' => array_values($channels),
+            'queries'  => array_values($queries),
+        ];
+    }
+
+    public function unsubscribe(string $subscriptionId): void
+    {
+        unset($this->subscriptions[$subscriptionId]);
+    }
+
+    /**
+     * Return subscription IDs whose channel set intersects the given channels.
+     *
+     * @param string[] $channels
+     * @return string[]
+     */
+    public function matchingSubscriptions(array $channels): array
+    {
+        $matches = [];
+        foreach ($this->subscriptions as $id => $sub) {
+            if (!empty(array_intersect($channels, $sub['channels']))) {
+                $matches[] = $id;
+            }
+        }
+        return $matches;
+    }
+}