Import products using Akeneo events

config/services.xml

@@ -61,6 +61,16 @@
             </call>
         </service>
 
+        <service id="webgriffe_sylius_akeneo.controller.webhook" class="Webgriffe\SyliusAkeneoPlugin\Controller\WebhookController" >
+            <tag name="controller.service_arguments"/>
+            <argument type="service" id="monolog.logger.webgriffe_sylius_akeneo_plugin" />
+            <argument type="service" id="webgriffe_sylius_akeneo.command_bus" />
+            <argument type="string">%webgriffe_sylius_akeneo.webhook.secret%</argument>
+            <call method="setContainer">
+                <argument type="service" id="service_container" />
+            </call>
+        </service>
+
         <service id="webgriffe_sylius_akeneo.temporary_file_manager" class="Webgriffe\SyliusAkeneoPlugin\TemporaryFilesManager">
             <argument type="service" id="filesystem" />
             <argument type="service">

config/webhook_routing.yaml

@@ -0,0 +1,4 @@
+webgriffe_sylius_akeneo_webhook:
+    path: /akeneo/webhook
+    methods: [POST]
+    controller: webgriffe_sylius_akeneo.controller.webhook::postAction

docs/Gemfile.lock

@@ -67,18 +67,21 @@ GEM
     sass-embedded (1.58.3)
       google-protobuf (~> 3.21)
       rake (>= 10.0.0)
+    sass-embedded (1.58.3-x86_64-darwin)
+      google-protobuf (~> 3.21)
     terminal-table (3.0.2)
       unicode-display_width (>= 1.1.1, < 3)
     unicode-display_width (2.4.2)
     webrick (1.8.1)
 
 PLATFORMS
   x86_64-darwin-22
+  x86_64-darwin-23
   x86_64-linux
 
 DEPENDENCIES
   jekyll (~> 4.3.2)
   just-the-docs
 
 BUNDLED WITH
-   2.4.19
+   2.4.22

docs/architecture_and_customization.md

@@ -1,20 +1,21 @@
 ---
 title: Architecture & customization
 layout: page
-nav_order: 5
+nav_order: 6
 ---
 
 # Architecture & customization
 
 > This plugin makes use of [Symfony Messenger](https://symfony.com/doc/current/messenger.html) component. It is highly
 > recommended to have a minimum knowledge of these component to understand how this integration works.
 
-This plugin has basically two entry points:
+This plugin has basically three entry points:
 
 * The UI admin import button, this will import only products
 * The Import CLI command, this will import both product, product associations and attribute options
+* The Webhook controller, this will import product and product associations when created/updated on Akeneo
 
-Both this entry points deals to identify entities to import from Akeneo. When they have collected them they dispatch
+These entry points deals to identify entities to import from Akeneo. When they have collected them they dispatch
 an `Webgriffe\SyliusAkeneoPlugin\Message\ItemImport` message on the messenger default bus.
 By default, in the configuration this message is handled by the main bus, the same bus used as default by Sylius for
 catalog promotions. This means that, if you have configured the main bus to run synchronously the import will be

docs/contributing.md

@@ -1,7 +1,7 @@
 ---
 title: Contributing
 layout: page
-nav_order: 6
+nav_order: 7
 ---
 
 # Contributing
@@ -175,6 +175,7 @@ WEBGRIFFE_SYLIUS_AKENEO_PLUGIN_CLIENT_ID=SAMPLE
 WEBGRIFFE_SYLIUS_AKENEO_PLUGIN_SECRET=SAMPLE
 WEBGRIFFE_SYLIUS_AKENEO_PLUGIN_USERNAME=SAMPLE
 WEBGRIFFE_SYLIUS_AKENEO_PLUGIN_PASSWORD=SAMPLE
+WEBGRIFFE_SYLIUS_AKENEO_PLUGIN_WEBHOOK_SECRET=WEBHOOK_SECRET
 ```
 
 Now, if you want you can import products from Akeneo to Sylius by launching the command:

docs/index.md

@@ -31,6 +31,7 @@ new features that this plugin could add. So, let's start! 🚀
 - [Installation](installation.html)
 - [Configuration](configuration.html)
 - [Usage](usage.html)
+- [Webhook](webhook.html)
 - [Architecture & customization](architecture_and_customization.html)
 - [Contributing](contributing.html)
 - [Upgrade guide](upgrade.html)

docs/requirements.md

@@ -9,7 +9,6 @@ nav_order: 1
 * PHP `^8.0`
 * Sylius `^1.12`
 * Symfony `^5.4` or `^6.0`
-* Akeneo PIM CE or EE `>= 3.2`.
-  The requirement for the version `3.2` is because the provided implementation of the product importer relies on
-  the `family_variant` key in the
-  Akeneo [GET Product model](https://api.akeneo.com/api-reference.html#get_product_models__code_) API response.
+* Akeneo PIM CE or EE `>= 5.0`.
+  The requirement for the version `5.0` is because the plugin now requires the Akeneo API events to work properly.
+  See https://api.akeneo.com/events-documentation/overview.html#welcome-to-the-events-api-basics-documentation

docs/upgrade/index.md

@@ -1,7 +1,7 @@
 ---
 title: Upgrade
 layout: page
-nav_order: 7
+nav_order: 8
 has_children: true
 ---
 

docs/upgrade/upgrade-2.0.md → docs/upgrade/upgrade-2.*.md

@@ -1,10 +1,15 @@
 ---
-title: Upgrade to 2.0
+title: Upgrade to 2.*
 layout: page
 nav_order: 0
 parent: Upgrade
 ---
 
+# Upgrade from `v2.2.0` to `v2.3.0`
+
+The v2.3.0 version introduces the support for webhooks. To enable check the new documentation [here](../webhook.html).
+It is highly recommended to remove the import command that runs every minute from your crontab and use the webhook instead.
+
 # Upgrade from `v1.17.0` to `v2.0.0`
 
 In the 2.0 version, we have introduced the Symfony Messenger component and removed all deprecations.

docs/usage.md

@@ -471,6 +471,9 @@ This will:
 * Import, every minute, all products that have been modified since the last execution, along with their associations
 * Reconcile Akeneo deleted products every 6 hours
 
+> *NB*: The line that imports products and product associations every minute should be added only if you do not use the
+> webhook feature (see next chapter). Otherwise, the products will be imported twice.
+
 Import and Reconcile commands uses a [lock mechanism](https://symfony.com/doc/current/console/lockable_trait.html) which
 prevents running them if another instance of the same command is already running.
 {% endraw %}

docs/webhook.md

@@ -0,0 +1,58 @@
+---
+title: Webhook
+layout: page
+nav_order: 5
+---
+
+{% raw %}
+
+# Webhook
+
+This plugin provides a webhook that can be used to automatically import products from Akeneo PIM to Sylius when they are
+created or updated.
+To use the webhook you need to:
+
+1. Import the routes needed for the plugin by adding the following to your `config/routes.yaml` file:
+    ```yaml
+    webgriffe_sylius_akeneo_plugin_webhook:
+        resource: "@WebgriffeSyliusAkeneoPlugin/config/webhook_routing.yaml"
+        prefix: ''
+    ```
+   The url of the webhook can be anything you want but it must be the same you will configure in Akeneo PIM. The
+   imported resource will use /akeneo/webhook, but if you prefer you can add any prefix you want or you can completely
+   rewrite the url:
+    ```yaml
+    webgriffe_sylius_akeneo_plugin_webhook:
+        path: /akeneo/complete/url/rewrite/webhook
+        methods: [POST]
+        controller: webgriffe_sylius_akeneo.controller.webhook::postAction
+    ```
+2. Configure the webhook in Akeneo PIM. Remember that events API are available from Akeneo 5. You can find the webhook
+   configuration in the Akeneo PIM's
+   menu: `Connect > Connection settings`. Select the current data destination connection (the one used from the plugin).
+   Now, select Event subscription from the left menu.
+   Check Event subscription activation and leave unchecked Use product UUID instead of product identifier? (this is not
+   currently supported). Now is time to insert the full URL previously configurated.
+   When you click the Save button, a new secret token will be generated. Copy it and paste it in the plugin's
+   configuration (see next step).
+   ![akeneo-event-subscrition.png](images%2Fakeneo-event-subscrition.png)
+3. In the plugin configuration (probably in the file config/packages/webgriffe_sylius_akeneo_plugin.yaml) add the
+   following:
+    ```yaml
+    webhook:
+        secret: 'YOUR_TOKEN_VALUE'
+    ```
+   Replace YOUR_TOKEN_VALUE with the secret token generated previously by Akeneo PIM. As always, we suggest to add this
+   token by using an env variable to keep it secret from the repository (
+   see [Symfony best practices doc](https://symfony.com/doc/current/best_practices.html#configuration)).
+4. If you want, you can now TEST the webhook with the dedicated button on Akeneo event subscription page. If any error
+   occurs, you can debug the webhook by adjusting the monolog.logger.webgriffe_sylius_akeneo_plugin monolog level to
+   debug, so that you will see if there is something that is currently not working.
+5. Finally, it is highly suggested that you remove the Product and ProductAssociations importer from the crontab to
+   avoid products imported twice:
+   ```diff
+    - *   *   *  *  *  /path/to/sylius/bin/console -e prod -q webgriffe:akeneo:import --since-file=/path/to/sylius/var/storage/akeneo-import-sincefile.txt --importer="Product" --importer="ProductAssociations"
+    ```
+
+{% endraw %}
+

src/Controller/WebhookController.php

@@ -0,0 +1,143 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Webgriffe\SyliusAkeneoPlugin\Controller;
+
+use const JSON_THROW_ON_ERROR;
+use Psr\Log\LoggerInterface;
+use RuntimeException;
+use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
+use Symfony\Component\HttpFoundation\Request;
+use Symfony\Component\HttpFoundation\Response;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Webgriffe\SyliusAkeneoPlugin\Message\ItemImport;
+use Webgriffe\SyliusAkeneoPlugin\Product\Importer as ProductImporter;
+use Webgriffe\SyliusAkeneoPlugin\ProductAssociations\Importer as ProductAssociationsImporter;
+
+/**
+ * @psalm-type AkeneoEventProduct = array{
+ *     uuid: string,
+ *     identifier: string,
+ *     enabled: bool,
+ *     family: string,
+ *     categories: string[],
+ *     groups: string[],
+ *     parent: ?string,
+ *     values: array<string, array>,
+ *     created: string,
+ *     updated: string,
+ *     associations: array<string, array>,
+ *     quantified_associations: array<string, array>,
+ * }
+ * @psalm-type AkeneoEventProductModel = array{
+ *     code: string,
+ *     family: string,
+ *     family_variant: string,
+ *     parent: ?string,
+ *     categories: string[],
+ *     values: array<string, array>,
+ *     created: string,
+ *     updated: string,
+ *     associations: array<string, array>,
+ *     quantified_associations: array<string, array>,
+ * }
+ * @psalm-type AkeneoEvent = array{
+ *     action: string,
+ *     event_id: string,
+ *     event_datetime: string,
+ *     author: string,
+ *     author_type: string,
+ *     pim_source: string,
+ *     data: array{
+ *         resource: AkeneoEventProduct|AkeneoEventProductModel
+ *     },
+ * }
+ * @psalm-type AkeneoEvents = array{
+ *     events: AkeneoEvent[],
+ * }
+ */
+final class WebhookController extends AbstractController
+{
+    public function __construct(
+        private LoggerInterface $logger,
+        private MessageBusInterface $messageBus,
+        private string $secret,
+    ) {
+    }
+
+    /**
+     * As guideline see the documentation here: https://api.akeneo.com/getting-started/quick-start-my-first-webhook-5x/step-2.html
+     *
+     * @throws RuntimeException
+     * @throws \JsonException
+     */
+    public function postAction(Request $request): Response
+    {
+        $timestamp = $request->headers->get('x-akeneo-request-timestamp');
+        $signature = $request->headers->get('x-akeneo-request-signature');
+        if (null === $timestamp || null === $signature) {
+            $this->logger->debug('The hash does not exists on the request! The request is not from Akeneo.');
+
+            return new Response('', Response::HTTP_UNAUTHORIZED);
+        }
+
+        /**
+         * @psalm-suppress UnnecessaryVarAnnotation
+         *
+         * @var string|resource $body on Symfony 5 the annotation is resource|string
+         */
+        $body = $request->getContent();
+        $expectedSignature = hash_hmac('sha256', $timestamp . '.' . (string) $body, $this->secret);
+        if (false === hash_equals($signature, $expectedSignature)) {
+            $this->logger->debug('The hash does not match! The request is not from Akeneo or the secret is wrong.');
+
+            return new Response('', Response::HTTP_UNAUTHORIZED);
+        }
+        if (time() - (int) $timestamp > 300) {
+            $this->logger->debug('The request is too old (> 5min)');
+
+            throw new RuntimeException('Request is too old (> 5min)');
+        }
+
+        if ($body === '') {
+            $this->logger->debug('The request body is empty, probably this request is a test from Event Subscription page on Akeneo.');
+
+            return new Response();
+        }
+
+        /**
+         * @TODO Could this be improved by using serializer? Is it necessary or overwork?
+         *
+         * @var AkeneoEvents $akeneoEvents
+         */
+        $akeneoEvents = json_decode((string) $body, true, 512, JSON_THROW_ON_ERROR);
+
+        foreach ($akeneoEvents['events'] as $akeneoEvent) {
+            $this->logger->debug(sprintf('Received event %s with id "%s"', $akeneoEvent['action'], $akeneoEvent['event_id']));
+
+            $resource = $akeneoEvent['data']['resource'];
+            if (array_key_exists('identifier', $resource)) {
+                $productCode = $resource['identifier'];
+                $this->logger->debug(sprintf(
+                    'Dispatching product import message for %s',
+                    $productCode,
+                ));
+                $this->messageBus->dispatch(new ItemImport(
+                    ProductImporter::AKENEO_ENTITY,
+                    $productCode,
+                ));
+                $this->logger->debug(sprintf(
+                    'Dispatching product associations import message for %s',
+                    $productCode,
+                ));
+                $this->messageBus->dispatch(new ItemImport(
+                    ProductAssociationsImporter::AKENEO_ENTITY,
+                    $productCode,
+                ));
+            }
+        }
+
+        return new Response();
+    }
+}

src/DependencyInjection/Configuration.php

@@ -33,6 +33,13 @@ public function getConfigTreeBuilder(): TreeBuilder
                     ->end()
                 ->end()
 
+                ->arrayNode('webhook')
+                    ->addDefaultsIfNotSet()
+                    ->children()
+                        ->scalarNode('secret')->isRequired()->cannotBeEmpty()->defaultNull()->end()
+                    ->end()
+                ->end()
+
                 ->arrayNode('value_handlers')
                     ->children()
                         ->arrayNode('product')

src/DependencyInjection/WebgriffeSyliusAkeneoExtension.php

@@ -115,13 +115,15 @@ final class WebgriffeSyliusAkeneoExtension extends AbstractResourceExtension imp
 
     public function load(array $configs, ContainerBuilder $container): void
     {
+        /** @var array{resources: array|mixed, api_client: array<array-key, ?string>, webhook: array{secret: ?string}, value_handlers: array} $config */
         $config = $this->processConfiguration($this->getConfiguration([], $container), $configs);
         $loader = new XmlFileLoader($container, new FileLocator(__DIR__ . '/../../config'));
 
         Assert::isArray($config['resources']);
         $this->registerResources('webgriffe_sylius_akeneo', 'doctrine/orm', $config['resources'], $container);
 
         $this->registerApiClientParameters($config['api_client'], $container);
+        $this->registerWebhookParameters($config['webhook'], $container);
 
         $loader->load('services.xml');
 
@@ -252,4 +254,12 @@ private function registerTemporaryDirectoryParameter(ContainerBuilder $container
         }
         $container->setParameter($parameterKey, sys_get_temp_dir());
     }
+
+    /**
+     * @param array{secret: ?string} $webhook
+     */
+    private function registerWebhookParameters(array $webhook, ContainerBuilder $container): void
+    {
+        $container->setParameter('webgriffe_sylius_akeneo.webhook.secret', $webhook['secret']);
+    }
 }

src/ProductAssociations/Importer.php

@@ -25,7 +25,7 @@
 
 final class Importer implements ImporterInterface
 {
-    private const AKENEO_ENTITY = 'ProductAssociations';
+    public const AKENEO_ENTITY = 'ProductAssociations';
 
     /**
      * @param RepositoryInterface<ProductAssociationInterface> $productAssociationRepository