diff --git a/app/realtime.php b/app/realtime.php index 5631a7f860..552823336f 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -394,15 +394,27 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, Console::success('Worker ' . $workerId . ' started successfully'); $telemetry = getTelemetry($workerId); + $realtimeDelayBuckets = [100, 250, 500, 750, 1000, 1500, 2000, 3000, 5000, 7500, 10000, 15000, 30000]; + $workerTelemetryAttributes = ['workerId' => (string) $workerId]; $register->set('telemetry', fn () => $telemetry); + $register->set('telemetry.workerAttributes', fn () => $workerTelemetryAttributes); + $register->set('telemetry.workerCounter', fn () => $telemetry->createUpDownCounter('realtime.server.active_workers')); + $register->set('telemetry.workerClientCounter', fn () => $telemetry->createUpDownCounter('realtime.server.worker_clients')); + $register->set('telemetry.workerSubscriptionCounter', fn () => $telemetry->createUpDownCounter('realtime.server.worker_subscriptions')); $register->set('telemetry.connectionCounter', fn () => $telemetry->createUpDownCounter('realtime.server.open_connections')); $register->set('telemetry.connectionCreatedCounter', fn () => $telemetry->createCounter('realtime.server.connection.created')); $register->set('telemetry.messageSentCounter', fn () => $telemetry->createCounter('realtime.server.message.sent')); $register->set('telemetry.deliveryDelayHistogram', fn () => $telemetry->createHistogram( name: 'realtime.server.delivery_delay', unit: 'ms', - advisory: ['ExplicitBucketBoundaries' => [100, 250, 500, 750, 1000, 1500, 2000, 3000, 5000, 7500, 10000, 15000, 30000]], + advisory: ['ExplicitBucketBoundaries' => $realtimeDelayBuckets], )); + $register->set('telemetry.arrivalDelayHistogram', fn () => $telemetry->createHistogram( + name: 'realtime.server.arrival_delay', + unit: 'ms', + advisory: ['ExplicitBucketBoundaries' => $realtimeDelayBuckets], + )); + $register->get('telemetry.workerCounter')->add(1); $attempts = 0; $start = time(); @@ -519,12 +531,28 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, $pubsub->subscribe(['realtime'], function (mixed $redis, string $channel, string $payload) use ($server, $workerId, $stats, $register, $realtime) { $event = json_decode($payload, true); + $eventTimestamp = $event['data']['timestamp'] ?? null; + if (\is_string($eventTimestamp)) { + try { + $eventDate = new \DateTimeImmutable($eventTimestamp, new \DateTimeZone('UTC')); + $now = new \DateTimeImmutable('now', new \DateTimeZone('UTC')); + $eventTimestampMs = (float) $eventDate->format('U.u') * 1000; + $nowTimestampMs = (float) $now->format('U.u') * 1000; + $arrivalDelayMs = (int) \max(0, $nowTimestampMs - $eventTimestampMs); + + $register->get('telemetry.arrivalDelayHistogram')->record($arrivalDelayMs); + } catch (\Throwable) { + // Ignore invalid timestamp payloads. + } + } + if ($event['permissionsChanged'] && isset($event['userId'])) { $projectId = $event['project']; $userId = $event['userId']; if ($realtime->hasSubscriber($projectId, 'user:' . $userId)) { $connection = array_key_first(reset($realtime->subscriptions[$projectId]['user:' . $userId])); + $subscriptionsBefore = \count($realtime->getSubscriptionMetadata($connection)); $consoleDatabase = getConsoleDB(); $project = $consoleDatabase->getAuthorization()->skip(fn () => $consoleDatabase->getDocument('projects', $projectId)); $database = getProjectDB($project); @@ -555,6 +583,12 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, if ($authorization !== null) { $realtime->connections[$connection]['authorization'] = $authorization; } + + $subscriptionsAfter = \count($realtime->getSubscriptionMetadata($connection)); + $subscriptionDelta = $subscriptionsAfter - $subscriptionsBefore; + if ($subscriptionDelta !== 0) { + $register->get('telemetry.workerSubscriptionCounter')->add($subscriptionDelta, $register->get('telemetry.workerAttributes')); + } } } @@ -600,7 +634,7 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, $updatedAt = $event['data']['payload']['$updatedAt'] ?? null; if (\is_string($updatedAt)) { try { - $updatedAtDate = new \DateTimeImmutable($updatedAt); + $updatedAtDate = new \DateTimeImmutable($updatedAt, new \DateTimeZone('UTC')); $now = new \DateTimeImmutable('now', new \DateTimeZone('UTC')); $updatedAtTimestampMs = (float) $updatedAtDate->format('U.u') * 1000; $nowTimestampMs = (float) $now->format('U.u') * 1000; @@ -640,6 +674,16 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, Console::error('Failed to restart pub/sub...'); }); +$server->onWorkerStop(function (int $workerId) use ($register) { + Console::warning('Worker ' . $workerId . ' stopping'); + + try { + $register->get('telemetry.workerCounter')->add(-1); + } catch (\Throwable $th) { + Console::error('Realtime onWorkerStop telemetry error: ' . $th->getMessage()); + } +}); + $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $register, $stats, &$realtime, $registerConnectionResources) { global $container; $request = new Request($request); @@ -728,6 +772,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $updateStats = static function (string $projectId, ?string $teamId, string $payloadJson) use ($register, $stats): void { $register->get('telemetry.connectionCounter')->add(1); + $register->get('telemetry.workerClientCounter')->add(1, $register->get('telemetry.workerAttributes')); $register->get('telemetry.connectionCreatedCounter')->add(1); $stats->set($projectId, [ @@ -792,6 +837,9 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $mapping[$index] = $subscriptionId; } + if (!empty($subscriptions)) { + $register->get('telemetry.workerSubscriptionCounter')->add(\count($subscriptions), $register->get('telemetry.workerAttributes')); + } $realtime->connections[$connection]['authorization'] = $authorization; @@ -846,7 +894,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, } }); -$server->onMessage(function (int $connection, string $message) use ($server, $realtime, $containerId) { +$server->onMessage(function (int $connection, string $message) use ($server, $realtime, $containerId, $register) { $project = null; $authorization = null; try { @@ -960,6 +1008,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $authorization = $realtime->connections[$connection]['authorization'] ?? null; $projectId = $realtime->connections[$connection]['projectId'] ?? null; + $subscriptionsBefore = \count($realtime->getSubscriptionMetadata($connection)); $meta = $realtime->getSubscriptionMetadata($connection); $realtime->unsubscribe($connection); @@ -984,6 +1033,12 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $realtime->connections[$connection]['authorization'] = $authorization; } + $subscriptionsAfter = \count($realtime->getSubscriptionMetadata($connection)); + $subscriptionDelta = $subscriptionsAfter - $subscriptionsBefore; + if ($subscriptionDelta !== 0) { + $register->get('telemetry.workerSubscriptionCounter')->add($subscriptionDelta, $register->get('telemetry.workerAttributes')); + } + $user = $response->output($user, Response::MODEL_ACCOUNT); $authResponsePayloadJson = json_encode([ @@ -1028,6 +1083,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re // bulk validation + parsing before subscribing $parsedPayloads = []; + $subscriptionsBefore = \count($realtime->getSubscriptionMetadata($connection)); foreach ($message['data'] as $payload) { if (!\is_array($payload)) { throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Each subscribe payload must be an object.'); @@ -1069,9 +1125,11 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $queries = $parsedPayload['queries']; $realtime->subscribe($projectId, $connection, $subscriptionId, $roles, $channels, $queries); } - - // subscribe() overwrites the connection entry; restore auth so later onMessage uses the same context. - $realtime->connections[$connection]['authorization'] = $authorization; + $subscriptionsAfter = \count($realtime->getSubscriptionMetadata($connection)); + $subscriptionDelta = $subscriptionsAfter - $subscriptionsBefore; + if ($subscriptionDelta !== 0) { + $register->get('telemetry.workerSubscriptionCounter')->add($subscriptionDelta, $register->get('telemetry.workerAttributes')); + } $responsePayload = json_encode([ 'type' => 'response', @@ -1102,6 +1160,65 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re break; + case 'unsubscribe': + if (!\is_array($message['data']) || !\array_is_list($message['data'])) { + throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Payload is not valid.'); + } + + $subscriptionsBefore = \count($realtime->getSubscriptionMetadata($connection)); + + // Validate every payload before executing any removal so an invalid entry + // later in the batch does not leave earlier entries half-applied on the server. + $validatedIds = []; + foreach ($message['data'] as $payload) { + if ( + !\is_array($payload) + || !\array_key_exists('subscriptionId', $payload) + || !\is_string($payload['subscriptionId']) + || $payload['subscriptionId'] === '' + ) { + throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Each unsubscribe payload must include a non-empty subscriptionId.'); + } + $validatedIds[] = $payload['subscriptionId']; + } + + $unsubscribeResults = []; + foreach ($validatedIds as $subscriptionId) { + $wasRemoved = $realtime->unsubscribeSubscription($connection, $subscriptionId); + $unsubscribeResults[] = [ + 'subscriptionId' => $subscriptionId, + 'removed' => $wasRemoved, + ]; + } + $subscriptionsAfter = \count($realtime->getSubscriptionMetadata($connection)); + $subscriptionDelta = $subscriptionsAfter - $subscriptionsBefore; + if ($subscriptionDelta !== 0) { + $register->get('telemetry.workerSubscriptionCounter')->add($subscriptionDelta, $register->get('telemetry.workerAttributes')); + } + + $unsubscribeResponsePayload = json_encode([ + 'type' => 'response', + 'data' => [ + 'to' => 'unsubscribe', + 'success' => true, + 'subscriptions' => $unsubscribeResults, + ], + ]); + + $server->send([$connection], $unsubscribeResponsePayload); + + if ($project !== null && !$project->isEmpty()) { + $unsubscribeOutboundBytes = \strlen($unsubscribeResponsePayload); + + if ($unsubscribeOutboundBytes > 0) { + triggerStats([ + METRIC_REALTIME_OUTBOUND => $unsubscribeOutboundBytes, + ], $project->getId()); + } + } + + break; + default: throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Message type is not valid.'); } @@ -1140,6 +1257,11 @@ $server->onClose(function (int $connection) use ($realtime, $stats, $register) { if (array_key_exists($connection, $realtime->connections)) { $stats->decr($realtime->connections[$connection]['projectId'], 'connectionsTotal'); $register->get('telemetry.connectionCounter')->add(-1); + $register->get('telemetry.workerClientCounter')->add(-1, $register->get('telemetry.workerAttributes')); + $subscriptionsBeforeClose = \count($realtime->getSubscriptionMetadata($connection)); + if ($subscriptionsBeforeClose > 0) { + $register->get('telemetry.workerSubscriptionCounter')->add(-$subscriptionsBeforeClose, $register->get('telemetry.workerAttributes')); + } $projectId = $realtime->connections[$connection]['projectId']; diff --git a/src/Appwrite/Messaging/Adapter/Realtime.php b/src/Appwrite/Messaging/Adapter/Realtime.php index f1d806bcc5..eeb1387674 100644 --- a/src/Appwrite/Messaging/Adapter/Realtime.php +++ b/src/Appwrite/Messaging/Adapter/Realtime.php @@ -114,14 +114,24 @@ class Realtime extends MessagingAdapter } } - // Keep userId from onOpen/authentication when provided. - // Fallback to existing stored value for subsequent subscribe upserts. - $this->connections[$identifier] = [ + // Union channels/roles across all subscriptions on the connection; overwriting would + // leave getSubscriptionMetadata and full unsubscribe operating on stale state. + $existing = $this->connections[$identifier] ?? []; + $existingChannels = $existing['channels'] ?? []; + $existingRoles = $existing['roles'] ?? []; + + $entry = [ 'projectId' => $projectId, - 'roles' => $roles, - 'userId' => $userId ?? ($this->connections[$identifier]['userId'] ?? ''), - 'channels' => $channels + 'roles' => \array_values(\array_unique(\array_merge($existingRoles, $roles))), + 'userId' => $userId ?? ($existing['userId'] ?? ''), + 'channels' => \array_values(\array_unique(\array_merge($existingChannels, $channels))), ]; + + if (\array_key_exists('authorization', $existing)) { + $entry['authorization'] = $existing['authorization']; + } + + $this->connections[$identifier] = $entry; } /** @@ -206,6 +216,87 @@ class Realtime extends MessagingAdapter } } + /** + * Removes a single subscription from a connection, keeping the connection alive so + * the client can resubscribe. Idempotent — returns true only when something was removed. + * + * @param mixed $connection + * @param string $subscriptionId + * @return bool + */ + public function unsubscribeSubscription(mixed $connection, string $subscriptionId): bool + { + $projectId = $this->connections[$connection]['projectId'] ?? ''; + if ($projectId === '' || !isset($this->subscriptions[$projectId])) { + return false; + } + + $removed = false; + + foreach ($this->subscriptions[$projectId] as $role => $byChannel) { + foreach ($byChannel as $channel => $byConnection) { + if (!isset($byConnection[$connection][$subscriptionId])) { + continue; + } + + unset($this->subscriptions[$projectId][$role][$channel][$connection][$subscriptionId]); + $removed = true; + + if (empty($this->subscriptions[$projectId][$role][$channel][$connection])) { + unset($this->subscriptions[$projectId][$role][$channel][$connection]); + } + if (empty($this->subscriptions[$projectId][$role][$channel])) { + unset($this->subscriptions[$projectId][$role][$channel]); + } + } + if (empty($this->subscriptions[$projectId][$role])) { + unset($this->subscriptions[$projectId][$role]); + } + } + + if (empty($this->subscriptions[$projectId])) { + unset($this->subscriptions[$projectId]); + } + + if ($removed) { + $this->recomputeConnectionState($connection); + } + + return $removed; + } + + /** + * Recomputes the cached channels on the connection entry from the subscriptions tree. + * Called after per-subscription removal so stale channel entries do not linger for later reads. + * + * Roles are deliberately NOT recomputed here. They represent the connection's authorization + * context (set at onOpen, replaced on `authentication` / permission-change) and must survive + * per-subscription removal — otherwise a client that unsubscribes every subscription and then + * resubscribes would subscribe with an empty roles array and silently receive nothing. + * + * @param mixed $connection + * @return void + */ + private function recomputeConnectionState(mixed $connection): void + { + if (!isset($this->connections[$connection])) { + return; + } + + $projectId = $this->connections[$connection]['projectId'] ?? ''; + $channels = []; + + foreach ($this->subscriptions[$projectId] ?? [] as $byChannel) { + foreach ($byChannel as $channel => $byConnection) { + if (isset($byConnection[$connection])) { + $channels[$channel] = true; + } + } + } + + $this->connections[$connection]['channels'] = \array_keys($channels); + } + /** * Checks if Channel has a subscriber. * @param string $projectId diff --git a/src/Appwrite/Platform/Modules/Functions/Workers/Screenshots.php b/src/Appwrite/Platform/Modules/Functions/Workers/Screenshots.php index 423bf0bd41..8fed46aa00 100644 --- a/src/Appwrite/Platform/Modules/Functions/Workers/Screenshots.php +++ b/src/Appwrite/Platform/Modules/Functions/Workers/Screenshots.php @@ -20,6 +20,8 @@ use Utopia\Platform\Action; use Utopia\Queue\Message; use Utopia\Storage\Device; use Utopia\System\System; +use Utopia\Telemetry\Adapter as Telemetry; +use Utopia\Telemetry\Counter; use function Swoole\Coroutine\batch; @@ -44,6 +46,7 @@ class Screenshots extends Action ->inject('dbForProject') ->inject('project') ->inject('deviceForFiles') + ->inject('telemetry') ->callback($this->action(...)); } @@ -53,7 +56,8 @@ class Screenshots extends Action Database $dbForPlatform, Database $dbForProject, Document $project, - Device $deviceForFiles + Device $deviceForFiles, + Telemetry $telemetry ): void { Console::log('Screenshot action started'); @@ -64,6 +68,7 @@ class Screenshots extends Action } $screenshotMessage = Screenshot::fromArray($payload); + $counter = $telemetry->createCounter('worker.screenshots.capture'); Console::log('Site screenshot started'); @@ -268,8 +273,24 @@ class Screenshots extends Action $date = \date('H:i:s'); $this->appendToLogs($dbForProject, $deployment->getId(), $queueForRealtime, "[$date] [appwrite] Screenshot capturing failed. Deployment will continue. \n"); + $this->recordTelemetry($counter, 'failure'); + throw $th; } + + $this->recordTelemetry($counter, 'success'); + } + + protected function recordTelemetry(Counter $counter, string $result): void + { + try { + $counter->add(1, [ + 'resourceType' => RESOURCE_TYPE_SITES, + 'result' => $result, + ]); + } catch (\Throwable) { + // Telemetry should never affect screenshot processing. + } } protected function appendToLogs(Database $dbForProject, string $deploymentId, Realtime $queueForRealtime, string $logs) diff --git a/src/Appwrite/Platform/Workers/Migrations.php b/src/Appwrite/Platform/Workers/Migrations.php index e69541820b..097597d126 100644 --- a/src/Appwrite/Platform/Workers/Migrations.php +++ b/src/Appwrite/Platform/Workers/Migrations.php @@ -195,9 +195,25 @@ class Migrations extends Action $migrationOptions = $migration->getAttribute('options'); /** @var Database|null $projectDB */ $projectDB = null; - if ($credentials['projectId']) { + $useAppwriteApiSource = false; + if ($source === SourceAppwrite::getName() && empty($credentials['projectId'])) { + throw new \Exception('Source projectId is required for Appwrite migrations'); + } + + if (! empty($credentials['projectId'])) { $this->sourceProject = $this->dbForPlatform->getDocument('projects', $credentials['projectId']); - $projectDB = call_user_func($this->getProjectDB, $this->sourceProject); + if ($this->sourceProject->isEmpty()) { + throw new \Exception('Source project not found for provided projectId'); + } + + $sourceRegion = $this->sourceProject->getAttribute('region', 'default'); + $destinationRegion = $this->project->getAttribute('region', 'default'); + $useAppwriteApiSource = $source === SourceAppwrite::getName() + && $destination === DestinationAppwrite::getName() + && $sourceRegion !== $destinationRegion; + if (! $useAppwriteApiSource) { + $projectDB = call_user_func($this->getProjectDB, $this->sourceProject); + } } $getDatabasesDB = fn (Document $database): Database => $this->getDatabasesDBForProject($database); @@ -233,7 +249,7 @@ class Migrations extends Action $credentials['endpoint'], $credentials['apiKey'], $getDatabasesDB, - SourceAppwrite::SOURCE_DATABASE, + $useAppwriteApiSource ? SourceAppwrite::SOURCE_API : SourceAppwrite::SOURCE_DATABASE, $projectDB, $queries ), @@ -580,9 +596,10 @@ class Migrations extends Action protected function getDatabasesDBForProject(Document $database) { - if ($this->sourceProject) { + if (isset($this->sourceProject) && ! $this->sourceProject->isEmpty()) { return ($this->getDatabasesDB)($database, $this->sourceProject); } + return ($this->getDatabasesDB)($database); } diff --git a/tests/e2e/Services/Realtime/RealtimeCustomClientQueryTestWithMessage.php b/tests/e2e/Services/Realtime/RealtimeCustomClientQueryTestWithMessage.php index edce428e0f..6376875157 100644 --- a/tests/e2e/Services/Realtime/RealtimeCustomClientQueryTestWithMessage.php +++ b/tests/e2e/Services/Realtime/RealtimeCustomClientQueryTestWithMessage.php @@ -164,6 +164,20 @@ class RealtimeCustomClientQueryTestWithMessage extends Scope return $response; } + /** + * @param array> $payloadEntries + * @return array + */ + private function sendUnsubscribeMessage(WebSocketClient $client, array $payloadEntries): array + { + $client->send(\json_encode([ + 'type' => 'unsubscribe', + 'data' => $payloadEntries, + ])); + + return \json_decode($client->receive(), true); + } + /** * subscriptionId: update with id from connected, create by omitting id, explicit new id, * duplicate id in one bulk (last wins), mixed bulk, idempotent repeat, empty queries → select-all. @@ -293,6 +307,282 @@ class RealtimeCustomClientQueryTestWithMessage extends Scope $client->close(); } + /** + * Update a subscription's queries/channels by reusing its subscriptionId. + * Verifies the update takes effect on live event filtering (not just the response echo), + * sibling subscriptions are untouched, unknown ids upsert as new, empty queries fall + * back to select-all, and a removed id can be recreated by subscribing again. + */ + public function testUpdateSubscriptionAndEdgeCases(): void + { + $user = $this->getUser(); + $userId = $user['$id'] ?? ''; + $session = $user['session'] ?? ''; + $projectId = $this->getProject()['$id']; + $headers = [ + 'origin' => 'http://localhost', + 'cookie' => 'a_session_' . $projectId . '=' . $session, + ]; + + $queryString = \http_build_query(['project' => $projectId]); + $client = new WebSocketClient( + 'ws://appwrite.test/v1/realtime?' . $queryString, + [ + 'headers' => $headers, + 'timeout' => 10, + ] + ); + $connected = \json_decode($client->receive(), true); + $this->assertEquals('connected', $connected['type'] ?? null); + + $triggerAccountEvent = function () use ($projectId, $session): void { + $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' => 'Update Sub Test ' . \uniqid()]); + }; + + // subA matches current user, subB never matches + $created = $this->sendSubscribeMessage($client, [ + [ + 'channels' => ['account'], + 'queries' => [Query::equal('$id', [$userId])->toString()], + ], + [ + 'channels' => ['account'], + 'queries' => [Query::equal('$id', ['no-match-initial'])->toString()], + ], + ]); + $subA = $created['data']['subscriptions'][0]['subscriptionId']; + $subB = $created['data']['subscriptions'][1]['subscriptionId']; + $this->assertNotSame($subA, $subB); + + $triggerAccountEvent(); + $event = \json_decode($client->receive(), true); + $this->assertEquals('event', $event['type']); + $this->assertSame([$subA], $event['data']['subscriptions']); + + // Swap: A -> non-matching, B -> matching. Same ids returned, server-side filter swaps. + $swap = $this->sendSubscribeMessage($client, [ + [ + 'subscriptionId' => $subA, + 'channels' => ['account'], + 'queries' => [Query::equal('$id', ['no-match-swapped'])->toString()], + ], + [ + 'subscriptionId' => $subB, + 'channels' => ['account'], + 'queries' => [Query::equal('$id', [$userId])->toString()], + ], + ]); + $this->assertSame($subA, $swap['data']['subscriptions'][0]['subscriptionId']); + $this->assertSame($subB, $swap['data']['subscriptions'][1]['subscriptionId']); + + $triggerAccountEvent(); + $event = \json_decode($client->receive(), true); + $this->assertEquals('event', $event['type']); + $this->assertSame([$subB], $event['data']['subscriptions']); + + // Sibling isolation: updating only subA must leave subB's matching filter intact. + $isolation = $this->sendSubscribeMessage($client, [[ + 'subscriptionId' => $subA, + 'channels' => ['account'], + 'queries' => [Query::equal('$id', [$userId])->toString()], + ]]); + $this->assertSame($subA, $isolation['data']['subscriptions'][0]['subscriptionId']); + + $triggerAccountEvent(); + $event = \json_decode($client->receive(), true); + $this->assertEquals('event', $event['type']); + $this->assertEqualsCanonicalizing([$subA, $subB], $event['data']['subscriptions']); + + // Empty queries on update -> select-all; subA still matches every event on the channel. + $empty = $this->sendSubscribeMessage($client, [[ + 'subscriptionId' => $subA, + 'channels' => ['account'], + 'queries' => [], + ]]); + $this->assertSame($subA, $empty['data']['subscriptions'][0]['subscriptionId']); + + $triggerAccountEvent(); + $event = \json_decode($client->receive(), true); + $this->assertEquals('event', $event['type']); + $this->assertEqualsCanonicalizing([$subA, $subB], $event['data']['subscriptions']); + + // Unknown subscriptionId upserts as a new subscription. + $ghostId = ID::unique(); + $ghost = $this->sendSubscribeMessage($client, [[ + 'subscriptionId' => $ghostId, + 'channels' => ['account'], + 'queries' => [Query::equal('$id', [$userId])->toString()], + ]]); + $this->assertSame($ghostId, $ghost['data']['subscriptions'][0]['subscriptionId']); + $this->assertNotSame($subA, $ghostId); + $this->assertNotSame($subB, $ghostId); + + $triggerAccountEvent(); + $event = \json_decode($client->receive(), true); + $this->assertEquals('event', $event['type']); + $this->assertEqualsCanonicalizing([$subA, $subB, $ghostId], $event['data']['subscriptions']); + + // Update after unsubscribe: subscribing with the removed id recreates it. + $unsub = $this->sendUnsubscribeMessage($client, [['subscriptionId' => $subA]]); + $this->assertTrue($unsub['data']['subscriptions'][0]['removed']); + + $triggerAccountEvent(); + $event = \json_decode($client->receive(), true); + $this->assertEquals('event', $event['type']); + $this->assertEqualsCanonicalizing([$subB, $ghostId], $event['data']['subscriptions']); + + $recreated = $this->sendSubscribeMessage($client, [[ + 'subscriptionId' => $subA, + 'channels' => ['account'], + 'queries' => [Query::equal('$id', [$userId])->toString()], + ]]); + $this->assertSame($subA, $recreated['data']['subscriptions'][0]['subscriptionId']); + + $triggerAccountEvent(); + $event = \json_decode($client->receive(), true); + $this->assertEquals('event', $event['type']); + $this->assertEqualsCanonicalizing([$subA, $subB, $ghostId], $event['data']['subscriptions']); + + $client->close(); + } + + public function testUnsubscribeRemovesOnlyMatchingSubscription(): void + { + $user = $this->getUser(); + $userId = $user['$id'] ?? ''; + $session = $user['session'] ?? ''; + $projectId = $this->getProject()['$id']; + $headers = [ + 'origin' => 'http://localhost', + 'cookie' => 'a_session_' . $projectId . '=' . $session, + ]; + + $queryString = \http_build_query(['project' => $projectId]); + $client = new WebSocketClient( + 'ws://appwrite.test/v1/realtime?' . $queryString, + [ + 'headers' => $headers, + 'timeout' => 10, + ] + ); + + $connected = \json_decode($client->receive(), true); + $this->assertEquals('connected', $connected['type'] ?? null); + + // Two subscriptions on the `account` channel, both matching the current user + $r1 = $this->sendSubscribeMessage($client, [[ + 'channels' => ['account'], + 'queries' => [Query::equal('$id', [$userId])->toString()], + ]]); + $subA = $r1['data']['subscriptions'][0]['subscriptionId']; + + $r2 = $this->sendSubscribeMessage($client, [[ + 'channels' => ['account'], + 'queries' => [Query::select(['*'])->toString()], + ]]); + $subB = $r2['data']['subscriptions'][0]['subscriptionId']; + + $this->assertNotSame($subA, $subB); + + // Trigger an event -- both subscriptions should match + $name = 'Unsubscribe Test ' . \uniqid(); + $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]); + + $event = \json_decode($client->receive(), true); + $this->assertEquals('event', $event['type']); + $this->assertEqualsCanonicalizing([$subA, $subB], $event['data']['subscriptions']); + + // Unsubscribe subA only + $unsubA = $this->sendUnsubscribeMessage($client, [['subscriptionId' => $subA]]); + $this->assertEquals('response', $unsubA['type']); + $this->assertEquals('unsubscribe', $unsubA['data']['to']); + $this->assertTrue($unsubA['data']['success']); + $this->assertCount(1, $unsubA['data']['subscriptions']); + $this->assertSame($subA, $unsubA['data']['subscriptions'][0]['subscriptionId']); + $this->assertTrue($unsubA['data']['subscriptions'][0]['removed']); + + // Trigger another event -- only subB should match now + $name = 'Unsubscribe Test ' . \uniqid(); + $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]); + + $event = \json_decode($client->receive(), true); + $this->assertEquals('event', $event['type']); + $this->assertSame([$subB], $event['data']['subscriptions']); + + // Idempotent: unsubscribing subA again reports removed=false + $unsubAgain = $this->sendUnsubscribeMessage($client, [['subscriptionId' => $subA]]); + $this->assertTrue($unsubAgain['data']['success']); + $this->assertFalse($unsubAgain['data']['subscriptions'][0]['removed']); + + // Connection is still alive -- ping still works + $client->send(\json_encode(['type' => 'ping'])); + $pong = \json_decode($client->receive(), true); + $this->assertEquals('pong', $pong['type']); + + // Invalid payloads are rejected + $errNonString = $this->sendUnsubscribeMessage($client, [['subscriptionId' => 123]]); + $this->assertEquals('error', $errNonString['type']); + $this->assertStringContainsString('subscriptionId', $errNonString['data']['message']); + + $errEmpty = $this->sendUnsubscribeMessage($client, [['subscriptionId' => '']]); + $this->assertEquals('error', $errEmpty['type']); + + $errMissing = $this->sendUnsubscribeMessage($client, [['channels' => ['foo']]]); + $this->assertEquals('error', $errMissing['type']); + + $errNonList = $this->sendUnsubscribeMessage($client, ['subscriptionId' => $subB]); + $this->assertEquals('error', $errNonList['type']); + + // A batch with a valid id followed by an invalid one must be rejected atomically: + // the valid id must remain subscribed, not be quietly removed before validation fails. + $partial = $this->sendUnsubscribeMessage($client, [ + ['subscriptionId' => $subB], + ['subscriptionId' => 999], + ]); + $this->assertEquals('error', $partial['type']); + + $name = 'Partial Rejection Test ' . \uniqid(); + $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]); + + $event = \json_decode($client->receive(), true); + $this->assertEquals('event', $event['type']); + $this->assertSame([$subB], $event['data']['subscriptions']); + + // Bulk unsubscribe: remaining subB plus a never-existed id -- response mirrors input order + $bulk = $this->sendUnsubscribeMessage($client, [ + ['subscriptionId' => $subB], + ['subscriptionId' => 'does-not-exist'], + ]); + $this->assertTrue($bulk['data']['success']); + $this->assertCount(2, $bulk['data']['subscriptions']); + $this->assertSame($subB, $bulk['data']['subscriptions'][0]['subscriptionId']); + $this->assertTrue($bulk['data']['subscriptions'][0]['removed']); + $this->assertSame('does-not-exist', $bulk['data']['subscriptions'][1]['subscriptionId']); + $this->assertFalse($bulk['data']['subscriptions'][1]['removed']); + + $client->close(); + } + public function testInvalidQueryShouldNotSubscribe(): void { $user = $this->getUser(); diff --git a/tests/unit/Messaging/MessagingTest.php b/tests/unit/Messaging/MessagingTest.php index 4b2474c760..f48be46202 100644 --- a/tests/unit/Messaging/MessagingTest.php +++ b/tests/unit/Messaging/MessagingTest.php @@ -147,6 +147,193 @@ class MessagingTest extends TestCase $this->assertEmpty($realtime->subscriptions); } + public function testSubscribeUnionsChannelsAndRoles(): void + { + $realtime = new Realtime(); + + $realtime->subscribe( + '1', + 1, + 'sub-a', + [Role::user(ID::custom('123'))->toString()], + ['documents'], + ); + + $realtime->subscribe( + '1', + 1, + 'sub-b', + [Role::users()->toString()], + ['files'], + ); + + $connection = $realtime->connections[1]; + + $this->assertContains('documents', $connection['channels']); + $this->assertContains('files', $connection['channels']); + $this->assertContains(Role::user(ID::custom('123'))->toString(), $connection['roles']); + $this->assertContains(Role::users()->toString(), $connection['roles']); + $this->assertCount(2, $connection['channels']); + $this->assertCount(2, $connection['roles']); + } + + public function testUnsubscribeSubscriptionRemovesOnlyOneSubscription(): void + { + $realtime = new Realtime(); + + $realtime->subscribe( + '1', + 1, + 'sub-a', + [Role::user(ID::custom('123'))->toString()], + ['documents'], + ); + + $realtime->subscribe( + '1', + 1, + 'sub-b', + [Role::users()->toString()], + ['files'], + ); + + $removed = $realtime->unsubscribeSubscription(1, 'sub-a'); + + $this->assertTrue($removed); + $this->assertArrayHasKey(1, $realtime->connections); + + // sub-a is fully cleaned from the tree + $this->assertArrayNotHasKey( + Role::user(ID::custom('123'))->toString(), + $realtime->subscriptions['1'] + ); + + // sub-b still delivers + $event = [ + 'project' => '1', + 'roles' => [Role::users()->toString()], + 'data' => [ + 'channels' => ['files'], + ], + ]; + $receivers = array_keys($realtime->getSubscribers($event)); + $this->assertEquals([1], $receivers); + + // Channels recomputed: sub-a's channel is gone + $this->assertSame(['files'], $realtime->connections[1]['channels']); + + // Roles are connection-level auth context — union of both subscribe calls preserved + $this->assertContains(Role::user(ID::custom('123'))->toString(), $realtime->connections[1]['roles']); + $this->assertContains(Role::users()->toString(), $realtime->connections[1]['roles']); + } + + public function testUnsubscribeSubscriptionIsIdempotent(): void + { + $realtime = new Realtime(); + + $realtime->subscribe( + '1', + 1, + 'sub-a', + [Role::users()->toString()], + ['documents'], + ); + + $this->assertFalse($realtime->unsubscribeSubscription(1, 'does-not-exist')); + $this->assertFalse($realtime->unsubscribeSubscription(99, 'sub-a')); + + // Original sub is untouched + $event = [ + 'project' => '1', + 'roles' => [Role::users()->toString()], + 'data' => [ + 'channels' => ['documents'], + ], + ]; + $this->assertEquals([1], array_keys($realtime->getSubscribers($event))); + } + + public function testUnsubscribeSubscriptionKeepsConnectionWhenLastSubRemoved(): void + { + $realtime = new Realtime(); + + $realtime->subscribe( + '1', + 1, + 'sub-a', + [Role::users()->toString()], + ['documents'], + ); + + $this->assertTrue($realtime->unsubscribeSubscription(1, 'sub-a')); + + $this->assertArrayHasKey(1, $realtime->connections); + $this->assertSame([], $realtime->connections[1]['channels']); + // Roles preserved so a later resubscribe on the same connection still has auth context + $this->assertSame([Role::users()->toString()], $realtime->connections[1]['roles']); + $this->assertArrayNotHasKey('1', $realtime->subscriptions); + } + + public function testResubscribeAfterUnsubscribingLastSubDelivers(): void + { + $realtime = new Realtime(); + + $realtime->subscribe( + '1', + 1, + 'sub-a', + [Role::users()->toString()], + ['documents'], + ); + + $this->assertTrue($realtime->unsubscribeSubscription(1, 'sub-a')); + + // Simulate the message-based subscribe path reading stored roles + $storedRoles = $realtime->connections[1]['roles']; + $this->assertNotEmpty($storedRoles, 'connection roles must survive per-subscription removal'); + + $realtime->subscribe('1', 1, 'sub-b', $storedRoles, ['files']); + + $event = [ + 'project' => '1', + 'roles' => [Role::users()->toString()], + 'data' => [ + 'channels' => ['files'], + ], + ]; + $this->assertEquals([1], array_keys($realtime->getSubscribers($event))); + } + + public function testSubscribeAfterOnOpenEmptySentinelPreservesUnion(): void + { + $realtime = new Realtime(); + + // Mirrors the onOpen empty-channels path: subscribe with '' id, empty channels + $realtime->subscribe( + '1', + 1, + '', + [Role::users()->toString()], + [], + [], + 'user-123', + ); + + // Now a real subscription comes in via the subscribe message type + $realtime->subscribe( + '1', + 1, + 'sub-a', + [Role::user(ID::custom('user-123'))->toString()], + ['documents'], + ); + + $this->assertSame('user-123', $realtime->connections[1]['userId']); + $this->assertContains('documents', $realtime->connections[1]['channels']); + $this->assertContains(Role::users()->toString(), $realtime->connections[1]['roles']); + $this->assertContains(Role::user(ID::custom('user-123'))->toString(), $realtime->connections[1]['roles']); + } + public function testConvertChannelsGuest(): void { $user = new Document([