feat(graphql): added support for graphql subscriptions to work for actions

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 @@
             $data[self::ITEM_IDENTIFIERS_KEY] = $this->identifiersExtractor->getIdentifiersFromItem($object, $context['operation'] ?? null);
         }
 
+        if (isset($context['graphql_operation_name']) && 'mercure_subscription' === $context['graphql_operation_name'] && \is_object($object) && isset($data['id']) && !isset($data['_id'])) {
+            $data['_id'] = $data['id'];
+            $data['id'] = $this->iriConverter->getIriFromResource($object);
+        }
+
         return $data;
     }
 
@@ -120,10 +127,46 @@
             return [...$attributeValue];
         }
 
+        // Handle relationships for mercure subscriptions
+        if ($operation instanceof QueryCollection && 'mercure_subscription' === $context['graphql_operation_name'] && $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
+            return $this->addPagination($attributeValue, $data, $context);
+        }
+
         // 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}
      */

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 @@
 
             $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;

src/GraphQl/Subscription/SubscriptionIdentifierGenerator.php

@@ -23,7 +23,21 @@
     public function generateSubscriptionIdentifier(array $fields): string
     {
         unset($fields['mercureUrl'], $fields['clientSubscriptionId']);
+        $fields = $this->removeTypename($fields);
 
         return hash('sha256', print_r($fields, true));
     }
+
+    private function removeTypename(array $data): array
+    {
+        foreach ($data as $key => $value) {
+            if ('__typename' === $key) {
+                unset($data[$key]);
+            } elseif (\is_array($value)) {
+                $data[$key] = $this->removeTypename($value);
+            }
+        }
+
+        return $data;
+    }
 }

src/GraphQl/Subscription/SubscriptionManager.php

@@ -47,43 +47,124 @@
         $fields = $info->getFieldSelection(\PHP_INT_MAX);
         $this->arrayRecursiveSort($fields, 'ksort');
         $iri = $operation ? $this->getIdentifierFromOperation($operation, $context['args'] ?? []) : $this->getIdentifierFromContext($context);
-        if (null === $iri) {
+        if (empty($iri)) {
             return null;
         }
+
+        $options = $operation ? ($operation->getMercure() ?? false) : false;
+        $private = $options['private'] ?? false;
+        $privateFields = $options['private_fields'] ?? [];
+        $previousObject = $context['graphql_context']['previous_object'] ?? null;
+        if ($private && $privateFields && $previousObject) {
+            foreach ($options['private_fields'] as $privateField) {
+                $fields['__private_field_'.$privateField] = $this->getResourceId($privateField, $previousObject);
+            }
+        }
+        if ($operation instanceof Subscription && $operation->isCollection()) {
+            $subscriptionId = $this->updateSubscriptionCollectionCacheData(
+                $iri,
+                $fields,
+            );
+        } else {
+            $subscriptionId = $this->updateSubscriptionItemCacheData(
+                $iri,
+                $fields,
+                $result,
+                $private,
+                $privateFields,
+                $previousObject
+            );
+        }
+
+        return $subscriptionId;
+    }
+
+    public function getPushPayloads(object $object, string $type): array
+    {
+        if ('delete' === $type) {
+            $payloads = $this->getDeletePushPayloads($object);
+        } else {
+            $payloads = $this->getCreatedOrUpdatedPayloads($object);
+        }
+
+        return $payloads;
+    }
+
+    /**
+     * @return array<array>
+     */
+    private function getSubscriptionsFromIri(string $iri): array
+    {
         $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;
-                }
-            }
+            return $subscriptionsCacheItem->get();
         }
 
-        $subscriptionId = $this->subscriptionIdentifierGenerator->generateSubscriptionIdentifier($fields);
-        unset($result['clientSubscriptionId']);
-        $subscriptions[] = [$subscriptionId, $fields, $result];
-        $subscriptionsCacheItem->set($subscriptions);
-        $this->subscriptionsCache->save($subscriptionsCacheItem);
+        return [];
+    }
 
-        return $subscriptionId;
+    private function removeItemFromSubscriptionCache(string $iri): void
+    {
+        $cacheKey = $this->encodeIriToCacheKey($iri);
+        if ($this->subscriptionsCache->hasItem($cacheKey)) {
+            $this->subscriptionsCache->deleteItem($cacheKey);
+        }
+    }
+
+    private function encodeIriToCacheKey(string $iri): string
+    {
+        return str_replace('/', '_', $iri);
     }
 
-    public function getPushPayloads(object $object): array
+    private function getResourceId(mixed $privateField, object $previousObject): string
+    {
+        $id = $previousObject->{'get'.ucfirst($privateField)}()->getId();
+        if ($id instanceof \Stringable) {
+            return (string) $id;
+        }
+
+        return $id;
+    }
+
+    private function getCollectionIri(string $iri): string
+    {
+        return substr($iri, 0, strrpos($iri, '/'));
+    }
+
+    private function getCreatedOrUpdatedPayloads(object $object): array
     {
         $iri = $this->iriConverter->getIriFromResource($object);
-        $subscriptions = $this->getSubscriptionsFromIri($iri);
+        // Add collection subscriptions
+        $subscriptions = array_merge(
+            $this->getSubscriptionsFromIri($this->getCollectionIri($iri)),
+            $this->getSubscriptionsFromIri($iri)
+        );
 
         $resourceClass = $this->getObjectClass($object);
         $resourceMetadata = $this->resourceMetadataCollectionFactory->create($resourceClass);
         $shortName = $resourceMetadata->getOperation()->getShortName();
 
+        $mercure = $resourceMetadata->getOperation()->getMercure() ?? false;
+        $private = $mercure['private'] ?? false;
+        $privateFieldsConfig = $mercure['private_fields'] ?? [];
+        $privateFieldData = [];
+        if ($private && $privateFieldsConfig) {
+            foreach ($privateFieldsConfig as $privateField) {
+                $privateFieldData['__private_field_'.$privateField] = $this->getResourceId($privateField, $object);
+            }
+        }
+
         $payloads = [];
         foreach ($subscriptions as [$subscriptionId, $subscriptionFields, $subscriptionResult]) {
+            if ($privateFieldData) {
+                $fieldDiff = array_intersect_assoc($subscriptionFields, $privateFieldData);
+                if ($fieldDiff !== $privateFieldData) {
+                    continue;
+                }
+            }
             $resolverContext = ['fields' => $subscriptionFields, 'is_collection' => false, 'is_mutation' => false, 'is_subscription' => true];
-            /** @var Operation */
-            $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']);
@@ -96,22 +177,80 @@
         return $payloads;
     }
 
-    /**
-     * @return array<array>
-     */
-    private function getSubscriptionsFromIri(string $iri): array
+    private function getDeletePushPayloads(object $object): array
     {
-        $subscriptionsCacheItem = $this->subscriptionsCache->getItem($this->encodeIriToCacheKey($iri));
+        $iri = $object->id;
+        $subscriptions = array_merge(
+            $this->getSubscriptionsFromIri($iri),
+            $this->getSubscriptionsFromIri($this->getCollectionIri($iri))
+        );
 
+        $payloads = [];
+        foreach ($subscriptions as [$subscriptionId, $subscriptionFields, $subscriptionResult]) {
+            $payloads[] = [$subscriptionId, ['type' => 'delete', 'payload' => $object]];
+        }
+        $this->removeItemFromSubscriptionCache($iri);
+
+        return $payloads;
+    }
+
+    private function updateSubscriptionItemCacheData(
+        string $iri,
+        array $fields,
+        ?array $result,
+        bool $private,
+        array $privateFields,
+        ?object $previousObject,
+    ): string {
+        $subscriptionsCacheItem = $this->subscriptionsCache->getItem($this->encodeIriToCacheKey($iri));
+        $subscriptions = [];
         if ($subscriptionsCacheItem->isHit()) {
-            return $subscriptionsCacheItem->get();
+            /*
+             * @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;
+                }
+            }
         }
 
-        return [];
+        $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 encodeIriToCacheKey(string $iri): string
-    {
-        return str_replace('/', '_', $iri);
+    private function updateSubscriptionCollectionCacheData(
+        string $iri,
+        array $fields,
+    ): string {
+        $subscriptionCollectionCacheItem = $this->subscriptionsCache->getItem(
+            $this->encodeIriToCacheKey($this->getCollectionIri($iri)),
+        );
+        $collectionSubscriptions = [];
+        if ($subscriptionCollectionCacheItem->isHit()) {
+            $collectionSubscriptions = $subscriptionCollectionCacheItem->get();
+            foreach ($collectionSubscriptions as [$subscriptionId, $subscriptionFields]) {
+                if ($subscriptionFields === $fields) {
+                    return $subscriptionId;
+                }
+            }
+        }
+        $subscriptionId = $this->subscriptionIdentifierGenerator->generateSubscriptionIdentifier($fields + ['__collection' => true]);
+        $collectionSubscriptions[] = [$subscriptionId, $fields];
+        $subscriptionCollectionCacheItem->set($collectionSubscriptions);
+        $this->subscriptionsCache->save($subscriptionCollectionCacheItem);
+
+        return $subscriptionId;
     }
 }

src/GraphQl/Subscription/SubscriptionManagerInterface.php

@@ -22,5 +22,5 @@ interface SubscriptionManagerInterface
 {
     public function retrieveSubscriptionId(array $context, ?array $result): ?string;
 
-    public function getPushPayloads(object $object): array;
+    public function getPushPayloads(object $object, string $type): array;
 }