feat(graphql): reworked how collection subscriptions work, added opt in mechanism

psihius · psihius · commit e764f2fbf19b · 2025-04-24T20:10:09.000+03:00
diff --git a/src/GraphQl/Serializer/ItemNormalizer.php b/src/GraphQl/Serializer/ItemNormalizer.php
@@ -16,6 +16,7 @@
 use ApiPlatform\GraphQl\State\Provider\NoopProvider;
 use ApiPlatform\Metadata\ApiProperty;
 use ApiPlatform\Metadata\GraphQl\Query;
+use ApiPlatform\Metadata\GraphQl\QueryCollection;
 use ApiPlatform\Metadata\IdentifiersExtractorInterface;
 use ApiPlatform\Metadata\IriConverterInterface;
 use ApiPlatform\Metadata\Property\Factory\PropertyMetadataFactoryInterface;
@@ -26,6 +27,7 @@
 use ApiPlatform\Metadata\Util\ClassInfoTrait;
 use ApiPlatform\Serializer\CacheKeyTrait;
 use ApiPlatform\Serializer\ItemNormalizer as BaseItemNormalizer;
+use Doctrine\Common\Collections\Collection;
 use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
 use Symfony\Component\PropertyAccess\PropertyAccessorInterface;
@@ -106,6 +108,11 @@ public function normalize(mixed $object, ?string $format = null, array $context
             $data[self::ITEM_IDENTIFIERS_KEY] = $this->identifiersExtractor->getIdentifiersFromItem($object, $context['operation'] ?? null);
         }
 
+        if ($context['graphql_operation_name'] === 'mercure_subscription' && is_object($object) && isset($data['id']) && !isset($data['_id'])) {
+            $data['_id'] = $data['id'];
+            $data['id'] = $this->iriConverter->getIriFromResource($object);
+        }
+
         return $data;
     }
 
@@ -120,10 +127,45 @@ protected function normalizeCollectionOfRelations(ApiProperty $propertyMetadata,
             return [...$attributeValue];
         }
 
+        // Handle relationships for mercure subscriptions
+        if ($operation instanceof QueryCollection && $context['graphql_operation_name'] === 'mercure_subscription' && $attributeValue instanceof Collection && !$attributeValue->isEmpty()) {
+            $relationContext = $context;
+            // Grab collection attributes
+            $relationContext['attributes'] = $context['attributes']['collection'];
+            // Iterate over the collection and normalize each item
+            $data['collection'] =  $attributeValue
+                ->map(fn($item) => $this->normalize($item, $format, $relationContext))
+                // Convert the collection to an array
+                ->toArray();
+            // Handle pagination if it's enabled in the query
+            $data = $this->addPagination($attributeValue, $data, $context);
+            return $data;
+        }
+
         // to-many are handled directly by the GraphQL resolver
         return [];
     }
 
+    private function addPagination(Collection $collection, array $data, array $context): array
+    {
+        if ($context['attributes']['paginationInfo'] ?? false) {
+            $data['paginationInfo'] = [];
+            if (array_key_exists('hasNextPage', $context['attributes']['paginationInfo'])) {
+                $data['paginationInfo']['hasNextPage'] = $collection->count() > ($context['pagination']['itemsPerPage'] ?? 10);
+            }
+            if (array_key_exists('itemsPerPage', $context['attributes']['paginationInfo'])) {
+                $data['paginationInfo']['itemsPerPage'] = $context['pagination']['itemsPerPage'] ?? 10;
+            }
+            if (array_key_exists('lastPage', $context['attributes']['paginationInfo'])) {
+                $data['paginationInfo']['lastPage'] = (int) ceil($collection->count() / ($context['pagination']['itemsPerPage'] ?? 10));
+            }
+            if (array_key_exists('totalCount', $context['attributes']['paginationInfo'])) {
+                $data['paginationInfo']['totalCount'] = $collection->count();
+            }
+        }
+        return $data;
+    }
+
     /**
      * {@inheritdoc}
      */
diff --git a/src/GraphQl/State/Processor/SubscriptionProcessor.php b/src/GraphQl/State/Processor/SubscriptionProcessor.php
@@ -17,6 +17,7 @@
 use ApiPlatform\GraphQl\Subscription\OperationAwareSubscriptionManagerInterface;
 use ApiPlatform\GraphQl\Subscription\SubscriptionManagerInterface;
 use ApiPlatform\Metadata\GraphQl\Operation as GraphQlOperation;
+use ApiPlatform\Metadata\GraphQl\Subscription;
 use ApiPlatform\Metadata\Operation;
 use ApiPlatform\State\ProcessorInterface;
 
@@ -49,6 +50,9 @@ public function process(mixed $data, Operation $operation, array $uriVariables =
 
             $hub = \is_array($mercure) ? ($mercure['hub'] ?? null) : null;
             $data['mercureUrl'] = $this->mercureSubscriptionIriGenerator->generateMercureUrl($subscriptionId, $hub);
+            if ($operation instanceof Subscription) {
+                $data['isCollection'] = $operation->isCollection();
+            }
         }
 
         return $data;
diff --git a/src/GraphQl/Subscription/SubscriptionManager.php b/src/GraphQl/Subscription/SubscriptionManager.php
@@ -50,6 +50,7 @@ public function retrieveSubscriptionId(array $context, ?array $result, ?Operatio
         if (empty($iri)) {
             return null;
         }
+
         $options = $operation->getMercure() ?? false;
         $private = $options['private'] ?? false;
         $privateFields = $options['private_fields'] ?? [];
@@ -59,33 +60,21 @@ public function retrieveSubscriptionId(array $context, ?array $result, ?Operatio
                 $fields['__private_field_'.$privateField] = $this->getResourceId($privateField, $previousObject);
             }
         }
-        $subscriptionsCacheItem = $this->subscriptionsCache->getItem($this->encodeIriToCacheKey($iri));
-        $subscriptions = [];
-        if ($subscriptionsCacheItem->isHit()) {
-            $subscriptions = $subscriptionsCacheItem->get();
-            foreach ($subscriptions as [$subscriptionId, $subscriptionFields, $subscriptionResult]) {
-                if ($subscriptionFields === $fields) {
-                    return $subscriptionId;
-                }
-            }
-        }
-
-        $subscriptionId = $this->subscriptionIdentifierGenerator->generateSubscriptionIdentifier($fields);
-        unset($result['clientSubscriptionId']);
-        if ($private && $privateFields && $previousObject) {
-            foreach ($options['private_fields'] as $privateField) {
-                unset($result['__private_field_'.$privateField]);
-            }
+        if ($operation->isCollection()) {
+            $subscriptionId = $this->updateSubscriptionCollectionCacheData(
+                $iri,
+                $fields,
+            );
+        } else {
+            $subscriptionId = $this->updateSubscriptionItemCacheData(
+                $iri,
+                $fields,
+                $result,
+                $private,
+                $privateFields,
+                $previousObject
+            );
         }
-        $subscriptions[] = [$subscriptionId, $fields, $result];
-        $subscriptionsCacheItem->set($subscriptions);
-        $this->subscriptionsCache->save($subscriptionsCacheItem);
-
-        $this->updateSubscriptionCollectionCacheData(
-            $iri,
-            $fields,
-            $subscriptions,
-        );
 
         return $subscriptionId;
     }
@@ -123,25 +112,9 @@ private function removeItemFromSubscriptionCache(string $iri): void
         }
     }
 
-    private function updateSubscriptionCollectionCacheData(
-        ?string                       $iri,
-        array                         $fields,
-        array                         $subscriptions,
-    ): void
+    private function encodeIriToCacheKey(string $iri): string
     {
-        $subscriptionCollectionCacheItem = $this->subscriptionsCache->getItem(
-            $this->encodeIriToCacheKey($this->getCollectionIri($iri)),
-        );
-        if ($subscriptionCollectionCacheItem->isHit()) {
-            $collectionSubscriptions = $subscriptionCollectionCacheItem->get();
-            foreach ($collectionSubscriptions as [$subscriptionId, $subscriptionFields, $subscriptionResult]) {
-                if ($subscriptionFields === $fields) {
-                    return;
-                }
-            }
-        }
-        $subscriptionCollectionCacheItem->set($subscriptions);
-        $this->subscriptionsCache->save($subscriptionCollectionCacheItem);
+        return str_replace('/', '_', $iri);
     }
 
     private function getResourceId(mixed $privateField, object $previousObject): string
@@ -161,11 +134,11 @@ private function getCollectionIri(string $iri): string
     private function getCreatedOrUpdatedPayloads(object $object): array
     {
         $iri = $this->iriConverter->getIriFromResource($object);
-        $subscriptions = $this->getSubscriptionsFromIri($iri);
-        if ($subscriptions === []) {
-            // Get subscriptions from collection Iri
-            $subscriptions = $this->getSubscriptionsFromIri($this->getCollectionIri($iri));
-        }
+        // Add collection subscriptions
+        $subscriptions = array_merge(
+            $this->getSubscriptionsFromIri($this->getCollectionIri($iri)),
+            $this->getSubscriptionsFromIri($iri)
+        );
 
         $resourceClass = $this->getObjectClass($object);
         $resourceMetadata = $this->resourceMetadataCollectionFactory->create($resourceClass);
@@ -190,7 +163,7 @@ private function getCreatedOrUpdatedPayloads(object $object): array
                 }
             }
             $resolverContext = ['fields' => $subscriptionFields, 'is_collection' => false, 'is_mutation' => false, 'is_subscription' => true];
-            $operation = (new Subscription())->withName('update_subscription')->withShortName($shortName);
+            $operation = (new Subscription())->withName('mercure_subscription')->withShortName($shortName);
             $data = $this->normalizeProcessor->process($object, $operation, [], $resolverContext);
 
             unset($data['clientSubscriptionId']);
@@ -219,8 +192,64 @@ private function getDeletePushPayloads(object $object): array
         return $payloads;
     }
 
-    private function encodeIriToCacheKey(string $iri): string
+    private function updateSubscriptionItemCacheData(
+        string  $iri,
+        array   $fields,
+        ?array  $result,
+        bool    $private,
+        array   $privateFields,
+        ?object $previousObject
+    ): string
     {
-        return str_replace('/', '_', $iri);
+        $subscriptionsCacheItem = $this->subscriptionsCache->getItem($this->encodeIriToCacheKey($iri));
+        $subscriptions = [];
+        if ($subscriptionsCacheItem->isHit()) {
+            /*
+             * @var array<array{string, array<string, string|array>, array<string, string|array>}>
+             */
+            $subscriptions = $subscriptionsCacheItem->get();
+            foreach ($subscriptions as [$subscriptionId, $subscriptionFields, $subscriptionResult]) {
+                if ($subscriptionFields === $fields) {
+                    return $subscriptionId;
+                }
+            }
+        }
+
+        $subscriptionId = $this->subscriptionIdentifierGenerator->generateSubscriptionIdentifier($fields);
+        unset($result['clientSubscriptionId']);
+        if ($private && $privateFields && $previousObject) {
+            foreach ($privateFields as $privateField) {
+                unset($result['__private_field_' . $privateField]);
+            }
+        }
+        $subscriptions[] = [$subscriptionId, $fields, $result];
+        $subscriptionsCacheItem->set($subscriptions);
+        $this->subscriptionsCache->save($subscriptionsCacheItem);
+        return $subscriptionId;
+    }
+
+
+
+    private function updateSubscriptionCollectionCacheData(
+        string $iri,
+        array  $fields,
+    ): string
+    {
+        $subscriptionCollectionCacheItem = $this->subscriptionsCache->getItem(
+            $this->encodeIriToCacheKey($this->getCollectionIri($iri)),
+        );
+        if ($subscriptionCollectionCacheItem->isHit()) {
+            $collectionSubscriptions = $subscriptionCollectionCacheItem->get();
+            foreach ($collectionSubscriptions as [$subscriptionId, $subscriptionFields]) {
+                if ($subscriptionFields === $fields) {
+                    return $subscriptionId;
+                }
+            }
+        }
+        $subscriptionId = $this->subscriptionIdentifierGenerator->generateSubscriptionIdentifier($fields + ['__collection' => true]);
+        $subscriptions[] = [$subscriptionId, $fields, []];
+        $subscriptionCollectionCacheItem->set($subscriptions);
+        $this->subscriptionsCache->save($subscriptionCollectionCacheItem);
+        return $subscriptionId;
     }
 }
diff --git a/src/GraphQl/Type/TypeBuilder.php b/src/GraphQl/Type/TypeBuilder.php
@@ -355,6 +355,7 @@ private function getResourceObjectTypeConfiguration(string $shortName, ResourceM
 
                     if ($operation instanceof Subscription) {
                         $fields['clientSubscriptionId'] = GraphQLType::string();
+                        $fields['isCollection'] = GraphQLType::boolean();
                         if ($operation->getMercure()) {
                             $fields['mercureUrl'] = GraphQLType::string();
                         }
diff --git a/src/Metadata/GraphQl/Subscription.php b/src/Metadata/GraphQl/Subscription.php

Original file line number	Diff line number	Diff line change
`@@ -355,6 +355,7 @@ private function getResourceObjectTypeConfiguration(string $shortName, ResourceM`
`355`	`355`
`356`	`356`	`if ($operation instanceof Subscription) {`
`357`	`357`	`$fields['clientSubscriptionId'] = GraphQLType::string();`
	`358`	`+ $fields['isCollection'] = GraphQLType::boolean();`
`358`	`359`	`if ($operation->getMercure()) {`
`359`	`360`	`$fields['mercureUrl'] = GraphQLType::string();`
`360`	`361`	`}`