Merge pull request #6 from aligent/feature/add-webhooks-and-oro-5.0-c…

…ompatibility Added Oro webhooks to the bundle and Oro 5.0 campatibility
aligent · Jan 17, 2023 · 756360b · 756360b
2 parents 0c2b1cb + ceae156
commit 756360b
Show file tree

Hide file tree

Showing 38 changed files with 1,816 additions and 139 deletions.
diff --git a/AligentAsyncBundle.php → AligentAsyncEventsBundle.php b/AligentAsyncBundle.php → AligentAsyncEventsBundle.php
@@ -10,11 +10,11 @@
  * @link      http://www.aligent.com.au/
  */
 
-namespace Aligent\AsyncBundle;
+namespace Aligent\AsyncEventsBundle;
 
 use Symfony\Component\HttpKernel\Bundle\Bundle;
 
-class AligentAsyncBundle extends Bundle
+class AligentAsyncEventsBundle extends Bundle
 {
 
-}
+}
diff --git a/Async/AbstractRetryableProcessor.php b/Async/AbstractRetryableProcessor.php
@@ -10,10 +10,10 @@
  * @link      http://www.aligent.com.au/
  */
 
-namespace Aligent\AsyncBundle\Async;
+namespace Aligent\AsyncEventsBundle\Async;
 
-use Aligent\AsyncBundle\Entity\FailedJob;
-use Aligent\AsyncBundle\Exception\RetryableException;
+use Aligent\AsyncEventsBundle\Entity\FailedJob;
+use Aligent\AsyncEventsBundle\Exception\RetryableException;
 use Doctrine\ORM\ORMException;
 use Oro\Component\MessageQueue\Consumption\MessageProcessorInterface;
 use Oro\Component\MessageQueue\Transport\MessageInterface;
@@ -63,7 +63,7 @@ public function process(MessageInterface $message, SessionInterface $session)
             $this->logger->error(
                 $e->getMessage(),
                 [
-                    'processor' => $message->getProperty(Config::PARAMETER_PROCESSOR_NAME),
+                    'topic' => $message->getProperty(Config::PARAMETER_TOPIC_NAME),
                     'headers'   => $message->getHeaders(),
                 ]
             );
@@ -114,4 +114,4 @@ protected function handleFailure(MessageInterface $message, RetryableException $
      * @throws RetryableException
      */
     abstract public function execute(MessageInterface $message);
-}
+}
diff --git a/Async/RetryableProcessorInterface.php b/Async/RetryableProcessorInterface.php
@@ -10,9 +10,9 @@
  * @link      http://www.aligent.com.au/
  */
 
-namespace Aligent\AsyncBundle\Async;
+namespace Aligent\AsyncEventsBundle\Async;
 
-use Aligent\AsyncBundle\Exception\RetryableException;
+use Aligent\AsyncEventsBundle\Exception\RetryableException;
 use Oro\Component\MessageQueue\Transport\MessageInterface;
 
 interface RetryableProcessorInterface
@@ -24,4 +24,4 @@ interface RetryableProcessorInterface
      * @return string
      */
     public function execute(MessageInterface $message);
-}
+}
diff --git a/Async/Topics.php b/Async/Topics.php
@@ -0,0 +1,27 @@
+<?php
+
+namespace Aligent\AsyncEventsBundle\Async;
+
+use Aligent\AsyncEventsBundle\Provider\WebhookConfigProvider;
+
+/**
+ * Class Topics
+ *
+ * @category  Aligent
+ * @package   Aligent\WebhookBundle\Async
+ * @author    Adam Hall <[email protected]>
+ * @copyright 2020 Aligent Consulting.
+ * @link      http://www.aligent.com.au/
+ */
+class Topics
+{
+    const WEBHOOK_ENTITY_CREATE = 'aligent.webhook.entity.create';
+    const WEBHOOK_ENTITY_UPDATE = 'aligent.webhook.entity.update';
+    const WEBHOOK_ENTITY_DELETE = 'aligent.webhook.entity.delete';
+
+    const EVENT_MAP = [
+        WebhookConfigProvider::UPDATE => self::WEBHOOK_ENTITY_UPDATE,
+        WebhookConfigProvider::DELETE => self::WEBHOOK_ENTITY_DELETE,
+        WebhookConfigProvider::CREATE => self::WEBHOOK_ENTITY_CREATE,
+    ];
+}
diff --git a/Async/WebhookEntityProcessor.php b/Async/WebhookEntityProcessor.php
@@ -0,0 +1,176 @@
+<?php
+
+namespace Aligent\AsyncEventsBundle\Async;
+
+use Aligent\AsyncEventsBundle\Exception\RetryableException;
+use Aligent\AsyncEventsBundle\Entity\WebhookTransport as WebhookTransportEntity;
+use Aligent\AsyncEventsBundle\Integration\WebhookTransport;
+use Aligent\AsyncEventsBundle\Provider\WebhookConfigProvider;
+use GuzzleHttp\Exception\GuzzleException;
+use Oro\Bundle\ImportExportBundle\Serializer\SerializerInterface;
+use Oro\Bundle\IntegrationBundle\Entity\Channel;
+use Oro\Component\MessageQueue\Client\Config;
+use Oro\Component\MessageQueue\Client\TopicSubscriberInterface;
+use Oro\Component\MessageQueue\Transport\MessageInterface;
+use Oro\Component\MessageQueue\Util\JSON;
+
+/**
+ * Class WebhookEntityHandler
+ *
+ * @category  Aligent
+ * @package   Aligent\WebhookBundle\Async
+ * @author    Adam Hall <[email protected]>
+ * @copyright 2020 Aligent Consulting.
+ * @link      http://www.aligent.com.au/
+ */
+class WebhookEntityProcessor extends AbstractRetryableProcessor implements TopicSubscriberInterface
+{
+    const EVENT_MAP = [
+        Topics::WEBHOOK_ENTITY_UPDATE => WebhookConfigProvider::UPDATE,
+        Topics::WEBHOOK_ENTITY_DELETE => WebhookConfigProvider::DELETE,
+        Topics::WEBHOOK_ENTITY_CREATE => WebhookConfigProvider::CREATE,
+    ];
+
+    /**
+     * @var WebhookConfigProvider
+     */
+    protected $configProvider;
+
+    /**
+     * @var WebhookTransport
+     */
+    protected $transport;
+
+    /**
+     * @var SerializerInterface
+     */
+    protected $serializer;
+
+    /**
+     * @param SerializerInterface $serializer
+     * @return WebhookEntityProcessor
+     */
+    public function setSerializer(SerializerInterface $serializer): WebhookEntityProcessor
+    {
+        $this->serializer = $serializer;
+
+        return $this;
+    }
+
+    /**
+     * @param WebhookConfigProvider $configProvider
+     * @return WebhookEntityProcessor
+     */
+    public function setConfigProvider(WebhookConfigProvider $configProvider): WebhookEntityProcessor
+    {
+        $this->configProvider = $configProvider;
+
+        return $this;
+    }
+
+    /**
+     * @param WebhookTransport $transport
+     * @return WebhookEntityProcessor
+     */
+    public function setTransport(WebhookTransport $transport): WebhookEntityProcessor
+    {
+        $this->transport = $transport;
+
+        return $this;
+    }
+
+    /**
+     * @inheritDoc
+     */
+    public function execute(MessageInterface $message)
+    {
+        $data = JSON::decode($message->getBody());
+        $topic = $message->getProperty(Config::PARAMETER_TOPIC_NAME);
+        $channelRepo = $this->registry->getRepository(Channel::class);
+        $channel = $channelRepo->find($data['channelId']);
+
+        if (!$channel) {
+            $this->logger->critical("Channel {$channel->getName()} no longer exists. Skipping webhook event.");
+            return self::REJECT;
+        }
+
+        try {
+            /** @var WebhookTransportEntity $transport */
+            $transport = $channel->getTransport();
+            $this->transport->init($transport);
+            $eventType = self::EVENT_MAP[$topic];
+
+            $response = $this->transport->sendWebhookEvent(
+                $transport->getMethod(),
+                $this->buildPayload($eventType, $data)
+            );
+            $this->logger->info(
+                'Webhook sent',
+                [
+                    'eventType' => $eventType,
+                    'message' => $data,
+                    'response' => $response
+                ]
+            );
+        } catch (\Exception $exception) {
+            throw new RetryableException($exception->getMessage(), $exception->getCode(), $exception);
+        } catch (GuzzleException $e) {
+            $message = "Server responded with non-200 status code";
+            $this->logger->error(
+                $message,
+                [
+                    'channelId' => $channel->getId(),
+                    'channel' => $channel->getName(),
+                    'topic' => $topic,
+                    'exception' => $e
+                ]
+            );
+            throw new RetryableException($message, 0, $e);
+        }
+
+        return self::ACK;
+    }
+
+    /**
+     * @inheritDoc
+     */
+    public static function getSubscribedTopics()
+    {
+        return [
+            Topics::WEBHOOK_ENTITY_CREATE,
+            Topics::WEBHOOK_ENTITY_DELETE,
+            Topics::WEBHOOK_ENTITY_UPDATE,
+        ];
+    }
+
+    /**
+     * @param string $event
+     * @param array $data
+     * @return array
+     * @throws \Symfony\Component\Serializer\Exception\ExceptionInterface
+     */
+    protected function buildPayload(string $event, array $data)
+    {
+        $entity = $this->registry->getRepository($data['class'])->find($data['id']);
+
+        if ($event === WebhookConfigProvider::CREATE) {
+            $changeSet = [];
+        } else {
+            // extract all of the before values from the change set
+            $changeSet = [];
+            foreach ($data['changeSet'] as $field => $changes) {
+                $changeSet[$field] = $changes[0];
+            }
+        }
+
+        $reflClass = new \ReflectionClass($data['class']);
+
+        return [
+            'type' => $reflClass->getShortName(),
+            'id' => count($data['id']) > 1 ? $data['id'] : reset($data['id']),
+            'operation' => $event,
+            'attributes' => $this->serializer->normalize($entity, null, ['webhook']),
+            'before' => $changeSet
+        ];
+    }
+}
diff --git a/Controller/FailedJobController.php b/Controller/FailedJobController.php
@@ -10,19 +10,19 @@
  * @link      http://www.aligent.com.au/
  */
 
-namespace Aligent\AsyncBundle\Controller;
+namespace Aligent\AsyncEventsBundle\Controller;
 
-use Aligent\AsyncBundle\Entity\FailedJob;
+use Aligent\AsyncEventsBundle\Entity\FailedJob;
 use Oro\Bundle\SecurityBundle\Annotation\AclAncestor;
 use Oro\Component\MessageQueue\Client\MessageProducerInterface;
 use Oro\Component\MessageQueue\Transport\Exception\Exception;
 use Sensio\Bundle\FrameworkExtraBundle\Configuration\Route;
 use Sensio\Bundle\FrameworkExtraBundle\Configuration\Template;
 use Oro\Bundle\SecurityBundle\Annotation\Acl;
-use Symfony\Bundle\FrameworkBundle\Controller\Controller;
+use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
 use Symfony\Component\HttpFoundation\JsonResponse;
 
-class FailedJobController extends Controller
+class FailedJobController extends AbstractController
 {
     /**
      * @Route(name="aligent_failed_jobs_index")
@@ -71,7 +71,6 @@ public function deleteAction(FailedJob $job)
             $em->flush();
         } catch (Exception $exception) {
             return new JsonResponse(['successful' => false]);
-
         }
 
         return new JsonResponse(['successful' => true]);
@@ -102,9 +101,8 @@ public function retryAction(FailedJob $job)
             $em->flush();
         } catch (Exception $exception) {
             return new JsonResponse(['successful' => false]);
-
         }
 
         return new JsonResponse(['successful' => true]);
     }
-}
+}
diff --git a/Datagrid/Extension/MassAction/MassRetryActionHandler.php b/Datagrid/Extension/MassAction/MassRetryActionHandler.php
@@ -10,10 +10,9 @@
  * @link      http://www.aligent.com.au/
  */
 
-namespace Aligent\AsyncBundle\Datagrid\Extension\MassAction;
+namespace Aligent\AsyncEventsBundle\Datagrid\Extension\MassAction;
 
-
-use Aligent\AsyncBundle\Entity\FailedJob;
+use Aligent\AsyncEventsBundle\Entity\FailedJob;
 use Oro\Bundle\DataGridBundle\Extension\MassAction\MassActionHandlerArgs;
 use Oro\Bundle\DataGridBundle\Extension\MassAction\MassActionHandlerInterface;
 use Oro\Bundle\DataGridBundle\Extension\MassAction\MassActionResponse;
@@ -72,4 +71,4 @@ public function handle(MassActionHandlerArgs $args)
             ? new MassActionResponse(true, "$count Jobs resent")
             : new MassActionResponse(false, "$count Jobs resent, $failed jobs failed to be resent");
     }
-}
+}
diff --git a/...ndencyInjection/AligentAsyncExtension.php → ...Injection/AligentAsyncEventsExtension.php b/...ndencyInjection/AligentAsyncExtension.php → ...Injection/AligentAsyncEventsExtension.php
@@ -10,15 +10,16 @@
  * @link      http://www.aligent.com.au/
  */
 
-namespace Aligent\AsyncBundle\DependencyInjection;
+namespace Aligent\AsyncEventsBundle\DependencyInjection;
 
 use Symfony\Component\Config\FileLocator;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
-use Symfony\Component\DependencyInjection\Extension\Extension;
+use Symfony\Component\HttpKernel\DependencyInjection\Extension;
 use Symfony\Component\DependencyInjection\Loader;
 
-class AligentAsyncExtension extends Extension
+class AligentAsyncEventsExtension extends Extension
 {
+
     /**
      * Loads a specific configuration.
      *
@@ -30,5 +31,6 @@ public function load(array $configs, ContainerBuilder $container)
     {
         $loader = new Loader\YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
         $loader->load('services.yml');
+        $loader->load('integration.yml');
     }
-}
+}
diff --git a/Entity/FailedJob.php b/Entity/FailedJob.php
@@ -10,15 +10,15 @@
  * @link      http://www.aligent.com.au/
  */
 
-namespace Aligent\AsyncBundle\Entity;
+namespace Aligent\AsyncEventsBundle\Entity;
 
 use Doctrine\ORM\Mapping as ORM;
 use Oro\Bundle\EntityBundle\EntityProperty\DatesAwareInterface;
 use Oro\Bundle\EntityBundle\EntityProperty\DatesAwareTrait;
 
 /**
  * Class FailedJob
- * @package Aligent\AsyncBundle\Entity
+ * @package Aligent\AsyncEventsBundle\Entity
  * @ORM\Entity()
  * @ORM\Table(
  *      name="aligent_failed_job",
@@ -75,8 +75,7 @@ public function __construct(
         $topic,
         array $body,
         \Exception $exception = null
-    )
-    {
+    ) {
         $this->topic = $topic;
         $this->body = $body;
 
@@ -181,4 +180,4 @@ public function setTrace($trace)
 
         return $this;
     }
-}
+}