From e0fec8f550b098d77192758419bef031cf62ceb1 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 15:42:17 +0530 Subject: [PATCH] updated --- app/realtime.php | 68 +++++++++++++++++++++++ src/Appwrite/SDK/Specification/Format.php | 7 +-- 2 files changed, 71 insertions(+), 4 deletions(-) diff --git a/app/realtime.php b/app/realtime.php index 3461ca83e5..13ebb609a8 100644 --- a/app/realtime.php +++ b/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); }); diff --git a/src/Appwrite/SDK/Specification/Format.php b/src/Appwrite/SDK/Specification/Format.php index 08f960b2a7..d48a4b8f3f 100644 --- a/src/Appwrite/SDK/Specification/Format.php +++ b/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';