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

psihius · psihius · commit cc46e696bdf7 · 2025-04-24T14:52:07.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;
@@ -120,10 +122,56 @@ 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(
+                    function($item) use ($format, $relationContext) {
+                        // Convert collection entity/item to array
+                        $normalized = $this->normalize($item, $format, $relationContext);
+                        // Add IRI to the normalized data to match GraphQL responses for ApiPlatform
+                        if (isset($normalized['id'])) {
+                            $normalized['_id'] = $normalized['id'];
+                            $normalized['id'] = $this->iriConverter->getIriFromResource($item);
+                        }
+                        return $normalized;
+                    }
+                )
+                // 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
@@ -76,63 +76,69 @@ public function __construct(
         mixed $rules = null,
         ?string $policy = null,
         array $extraProperties = [],
+        protected bool $collection = false,
     ) {
         parent::__construct(
-            resolver: $resolver,
-            args: $args,
-            extraArgs: $extraArgs,
-            links: $links,
-            securityAfterResolver: $securityAfterResolver,
-            securityMessageAfterResolver: $securityMessageAfterResolver,
-            shortName: $shortName,
-            class: $class,
-            paginationEnabled: $paginationEnabled,
-            paginationType: $paginationType,
-            paginationItemsPerPage: $paginationItemsPerPage,
-            paginationMaximumItemsPerPage: $paginationMaximumItemsPerPage,
-            paginationPartial: $paginationPartial,
-            paginationClientEnabled: $paginationClientEnabled,
-            paginationClientItemsPerPage: $paginationClientItemsPerPage,
-            paginationClientPartial: $paginationClientPartial,
-            paginationFetchJoinCollection: $paginationFetchJoinCollection,
-            paginationUseOutputWalkers: $paginationUseOutputWalkers,
-            order: $order,
-            description: $description,
-            normalizationContext: $normalizationContext,
-            denormalizationContext: $denormalizationContext,
-            collectDenormalizationErrors: $collectDenormalizationErrors,
-            security: $security,
-            securityMessage: $securityMessage,
-            securityPostDenormalize: $securityPostDenormalize,
-            securityPostDenormalizeMessage: $securityPostDenormalizeMessage,
-            securityPostValidation: $securityPostValidation,
-            securityPostValidationMessage: $securityPostValidationMessage,
-            deprecationReason: $deprecationReason,
-            filters: $filters,
-            validationContext: $validationContext,
-            input: $input,
-            output: $output,
-            mercure: $mercure,
-            messenger: $messenger,
-            elasticsearch: $elasticsearch,
-            urlGenerationStrategy: $urlGenerationStrategy,
-            read: $read,
-            deserialize: $deserialize,
-            validate: $validate,
-            write: $write,
-            serialize: $serialize,
-            fetchPartial: $fetchPartial,
-            forceEager: $forceEager,
-            priority: $priority,
-            name: $name ?: 'update_subscription',
-            provider: $provider,
-            processor: $processor,
-            stateOptions: $stateOptions,
-            parameters: $parameters,
+            resolver                       : $resolver,
+            args                           : $args,
+            extraArgs                      : $extraArgs,
+            links                          : $links,
+            securityAfterResolver          : $securityAfterResolver,
+            securityMessageAfterResolver   : $securityMessageAfterResolver,
+            shortName                      : $shortName,
+            class                          : $class,
+            paginationEnabled              : $paginationEnabled,
+            paginationType                 : $paginationType,
+            paginationItemsPerPage         : $paginationItemsPerPage,
+            paginationMaximumItemsPerPage  : $paginationMaximumItemsPerPage,
+            paginationPartial              : $paginationPartial,
+            paginationClientEnabled        : $paginationClientEnabled,
+            paginationClientItemsPerPage   : $paginationClientItemsPerPage,
+            paginationClientPartial        : $paginationClientPartial,
+            paginationFetchJoinCollection  : $paginationFetchJoinCollection,
+            paginationUseOutputWalkers     : $paginationUseOutputWalkers,
+            order                          : $order,
+            description                    : $description,
+            normalizationContext           : $normalizationContext,
+            denormalizationContext         : $denormalizationContext,
+            collectDenormalizationErrors   : $collectDenormalizationErrors,
+            security                       : $security,
+            securityMessage                : $securityMessage,
+            securityPostDenormalize        : $securityPostDenormalize,
+            securityPostDenormalizeMessage : $securityPostDenormalizeMessage,
+            securityPostValidation         : $securityPostValidation,
+            securityPostValidationMessage  : $securityPostValidationMessage,
+            deprecationReason              : $deprecationReason,
+            filters                        : $filters,
+            validationContext              : $validationContext,
+            input                          : $input,
+            output                         : $output,
+            mercure                        : $mercure,
+            messenger                      : $messenger,
+            elasticsearch                  : $elasticsearch,
+            urlGenerationStrategy          : $urlGenerationStrategy,
+            read                           : $read,
+            deserialize                    : $deserialize,
+            validate                       : $validate,
+            write                          : $write,
+            serialize                      : $serialize,
+            fetchPartial                   : $fetchPartial,
+            forceEager                     : $forceEager,
+            priority                       : $priority,
+            name                           : $name ?: 'update_subscription',
+            provider                       : $provider,
+            processor                      : $processor,
+            stateOptions                   : $stateOptions,
+            parameters                     : $parameters,
             queryParameterValidationEnabled: $queryParameterValidationEnabled,
-            policy: $policy,
-            rules: $rules,
-            extraProperties: $extraProperties,
+            rules                          : $rules,
+            policy                         : $policy,
+            extraProperties                : $extraProperties,
         );
     }
+
+    public function isCollection(): bool
+    {
+        return $this->collection;
+    }
 }

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`	`}`