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

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 ($key === '__typename') {
+                unset($data[$key]);
+            } elseif (is_array($value)) {
+                $data[$key] = $this->removeTypename($value);
+            }
+        }
+
+        return $data;
+    }
 }

src/GraphQl/Subscription/SubscriptionManager.php

@@ -47,9 +47,18 @@
         $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->getMercure() ?? 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);
+            }
+        }
         $subscriptionsCacheItem = $this->subscriptionsCache->getItem($this->encodeIriToCacheKey($iri));
         $subscriptions = [];
         if ($subscriptionsCacheItem->isHit()) {
@@ -63,26 +72,124 @@
 
         $subscriptionId = $this->subscriptionIdentifierGenerator->generateSubscriptionIdentifier($fields);
         unset($result['clientSubscriptionId']);
+        if ($private && $privateFields && $previousObject) {
+            foreach ($options['private_fields'] as $privateField) {
+                unset($result['__private_field_'.$privateField]);
+            }
+        }
         $subscriptions[] = [$subscriptionId, $fields, $result];
         $subscriptionsCacheItem->set($subscriptions);
         $this->subscriptionsCache->save($subscriptionsCacheItem);
 
+        $this->updateSubscriptionCollectionCacheData(
+            $iri,
+            $fields,
+            $subscriptions,
+        );
+
         return $subscriptionId;
     }
 
-    public function getPushPayloads(object $object): array
+    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));
+
+        if ($subscriptionsCacheItem->isHit()) {
+            return $subscriptionsCacheItem->get();
+        }
+
+        return [];
+    }
+
+    private function removeItemFromSubscriptionCache(string $iri): void
+    {
+        $cacheKey = $this->encodeIriToCacheKey($iri);
+        if ($this->subscriptionsCache->hasItem($cacheKey)) {
+            $this->subscriptionsCache->deleteItem($cacheKey);
+        }
+    }
+
+    private function updateSubscriptionCollectionCacheData(
+        ?string                       $iri,
+        array                         $fields,
+        array                         $subscriptions,
+    ): void
+    {
+        $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);
+    }
+
+    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);
+        if ($subscriptions === []) {
+            // Get subscriptions from collection Iri
+            $subscriptions = $this->getSubscriptionsFromIri($this->getCollectionIri($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);
             $data = $this->normalizeProcessor->process($object, $operation, [], $resolverContext);
 
@@ -92,22 +199,24 @@
                 $payloads[] = [$subscriptionId, $data];
             }
         }
-
         return $payloads;
     }
 
-    /**
-     * @return array<array>
-     */
-    private function getSubscriptionsFromIri(string $iri): array
+    private function getDeletePushPayloads(object $object): array
     {
-        $subscriptionsCacheItem = $this->subscriptionsCache->getItem($this->encodeIriToCacheKey($iri));
-
-        if ($subscriptionsCacheItem->isHit()) {
-            return $subscriptionsCacheItem->get();
+        $iri = $object->id;
+        $subscriptions = $this->getSubscriptionsFromIri($iri);
+        if ($subscriptions === []) {
+            // Get subscriptions from collection Iri
+            $subscriptions = $this->getSubscriptionsFromIri($this->getCollectionIri($iri));
         }
 
-        return [];
+        $payloads = [];
+        foreach ($subscriptions as [$subscriptionId, $subscriptionFields, $subscriptionResult]) {
+            $payloads[] = [$subscriptionId, ['type' => 'delete', 'payload' => $object]];
+        }
+        $this->removeItemFromSubscriptionCache($iri);
+        return $payloads;
     }
 
     private function encodeIriToCacheKey(string $iri): string

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

src/Symfony/Doctrine/EventListener/PublishMercureUpdatesListener.php

@@ -50,6 +50,7 @@ final class PublishMercureUpdatesListener
         'topics' => true,
         'data' => true,
         'private' => true,
+        'private_fields' => true,
         'id' => true,
         'type' => true,
         'retry' => true,
@@ -293,11 +294,11 @@ private function evaluateTopics(array &$options, object $object): void
      */
     private function getGraphQlSubscriptionUpdates(object $object, array $options, string $type): array
     {
-        if ('update' !== $type || !$this->graphQlSubscriptionManager || !$this->graphQlMercureSubscriptionIriGenerator) {
+        if (!$this->graphQlSubscriptionManager || !$this->graphQlMercureSubscriptionIriGenerator) {
             return [];
         }
 
-        $payloads = $this->graphQlSubscriptionManager->getPushPayloads($object);
+        $payloads = $this->graphQlSubscriptionManager->getPushPayloads($object, $type);
 
         $updates = [];
         foreach ($payloads as [$subscriptionId, $data]) {