updated

app/realtime.php

@@ -38,6 +38,7 @@ use Utopia\DSN\DSN;
 use Utopia\Logger\Log;
 use Utopia\Pools\Group;
 use Utopia\Registry\Registry;
+use Utopia\Span\Span;
 use Utopia\System\System;
 use Utopia\Telemetry\Adapter\None as NoTelemetry;
 use Utopia\WebSocket\Adapter;
@@ -326,11 +327,32 @@ if (!function_exists('logError')) {
     }
 }
 
+if (!function_exists('traceOperationalEvent')) {
+    function traceOperationalEvent(string $action, string $message, array $context = []): void
+    {
+        Span::init($action);
+        Span::add('realtime.action', $action);
+        Span::add('realtime.message', $message);
+        Span::add('realtime.timestamp', DateTime::formatTz(DateTime::now()));
+
+        foreach ($context as $key => $value) {
+            if (\is_scalar($value) || $value === null) {
+                Span::add('realtime.' . $key, ($value === null || $value === '') ? 'n/a' : $value);
+            }
+        }
+
+        Span::current()?->finish();
+    }
+}
+
 $server->error(logError(...));
 
 $server->onStart(function () use ($stats, $containerId, &$statsDocument) {
     sleep(5); // wait for the initial database schema to be ready
     Console::success('Server started successfully');
+    traceOperationalEvent('realtime.server.started', 'Realtime server started', [
+        'container' => $containerId,
+    ]);
 
     /**
      * Create document for this worker to share stats across Containers.
@@ -394,6 +416,9 @@ $server->onStart(function () use ($stats, $containerId, &$statsDocument) {
 
 $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, $realtime) {
     Console::success('Worker ' . $workerId . ' started successfully');
+    traceOperationalEvent('realtime.worker.started', 'Realtime worker started', [
+        'workerId' => $workerId,
+    ]);
 
     $telemetry = getTelemetry($workerId);
     $realtimeDelayBuckets = [100, 250, 500, 750, 1000, 1500, 2000, 3000, 5000, 7500, 10000, 15000, 30000];
@@ -526,6 +551,9 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats,
             if ($pubsub->ping(true)) {
                 $attempts = 0;
                 Console::success('Pub/sub connection established (worker: ' . $workerId . ')');
+                traceOperationalEvent('realtime.pubsub.connected', 'Realtime pubsub connected', [
+                    'workerId' => $workerId,
+                ]);
             } else {
                 Console::error('Pub/sub failed (worker: ' . $workerId . ')');
             }
@@ -858,6 +886,14 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
 
         $server->send([$connection], $connectedPayloadJson);
         $updateStats($project->getId(), $project->getAttribute('teamId'), $connectedPayloadJson);
+        traceOperationalEvent('realtime.connection.opened', 'Realtime connection established', [
+            'connectionId' => $connection,
+            'projectId' => $project->getId(),
+            'teamId' => $project->getAttribute('teamId'),
+            'userId' => $logUser?->getId() ?: null,
+            'channelCount' => \count($names),
+            'subscriptionCount' => \count($mapping),
+        ]);
 
 
     } catch (Throwable $th) {
@@ -1053,6 +1089,12 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
                 ]);
 
                 $server->send([$connection], $authResponsePayloadJson);
+                traceOperationalEvent('realtime.authentication.succeeded', 'Realtime authentication succeeded', [
+                    'connectionId' => $connection,
+                    'projectId' => $projectId,
+                    'userId' => $user['$id'] ?? null,
+                    'subscriptionDelta' => $subscriptionDelta,
+                ]);
 
                 if ($project !== null && !$project->isEmpty()) {
                     $authOutboundBytes = \strlen($authResponsePayloadJson);
@@ -1149,6 +1191,12 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
                 ]);
 
                 $server->send([$connection], $responsePayload);
+                traceOperationalEvent('realtime.subscribe.updated', 'Realtime subscriptions updated', [
+                    'connectionId' => $connection,
+                    'projectId' => $projectId,
+                    'subscriptionCount' => \count($parsedPayloads),
+                    'subscriptionDelta' => $subscriptionDelta,
+                ]);
 
                 if ($project !== null && !$project->isEmpty()) {
                     $subscribeOutboundBytes = \strlen($responsePayload);
@@ -1208,6 +1256,13 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
                 ]);
 
                 $server->send([$connection], $unsubscribeResponsePayload);
+                traceOperationalEvent('realtime.unsubscribe.updated', 'Realtime subscriptions removed', [
+                    'connectionId' => $connection,
+                    'projectId' => $projectId,
+                    'requestedCount' => \count($validatedIds),
+                    'removedCount' => \count(\array_filter($unsubscribeResults, fn (array $item) => $item['removed'] ?? false)),
+                    'subscriptionDelta' => $subscriptionDelta,
+                ]);
 
                 if ($project !== null && !$project->isEmpty()) {
                     $unsubscribeOutboundBytes = \strlen($unsubscribeResponsePayload);
@@ -1255,6 +1310,14 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
 });
 
 $server->onClose(function (int $connection) use ($realtime, $stats, $register) {
+    $projectId = null;
+    $userId = null;
+
+    if (array_key_exists($connection, $realtime->connections)) {
+        $projectId = $realtime->connections[$connection]['projectId'] ?? null;
+        $userId = $realtime->connections[$connection]['userId'] ?? null;
+    }
+
     try {
         if (array_key_exists($connection, $realtime->connections)) {
             $stats->decr($realtime->connections[$connection]['projectId'], 'connectionsTotal');
@@ -1278,6 +1341,11 @@ $server->onClose(function (int $connection) use ($realtime, $stats, $register) {
     }
     $realtime->unsubscribe($connection);
 
+    traceOperationalEvent('realtime.connection.closed', 'Realtime connection closed', [
+        'connectionId' => $connection,
+        'projectId' => $projectId,
+        'userId' => $userId,
+    ]);
     Console::info('Connection close: ' . $connection);
 });
 

src/Appwrite/SDK/Specification/Format.php

@@ -742,6 +742,7 @@ abstract class Format
                 }
                 break;
             case 'project':
+            case 'projects':
                 switch ($method) {
                     case 'getUsage':
                         switch ($param) {
@@ -749,10 +750,6 @@ abstract class Format
                                 return 'ProjectUsageRange';
                         }
                         break;
-                }
-                break;
-            case 'projects':
-                switch ($method) {
                     case 'getEmailTemplate':
                     case 'updateEmailTemplate':
                     case 'deleteEmailTemplate':
@@ -770,7 +767,9 @@ abstract class Format
                         }
                         break;
                     case 'createSmtpTest':
+                    case 'createSMTPTest':
                     case 'updateSmtp':
+                    case 'updateSMTP':
                         switch ($param) {
                             case 'secure':
                                 return 'SMTPSecure';