From 4dac01b164b06b02e1924b2974a071958db84f98 Mon Sep 17 00:00:00 2001
From: Manuele Menozzi <manuele.menozzi@gmail.com>
Date: Mon, 28 Oct 2024 14:53:19 +0100
Subject: [PATCH] Allow to configure ElasticSearch index in esb.yml

---
 esb.yml.sample                |  5 +++++
 src/FlowConfiguration.php     |  1 +
 src/Model/FlowConfig.php      |  8 ++++++++
 src/Service/ElasticSearch.php |  8 ++++++++
 src/Service/QueueManager.php  | 13 +++++++++++++
 5 files changed, 35 insertions(+)

diff --git a/esb.yml.sample b/esb.yml.sample
index 432348b..67e5ef6 100644
--- a/esb.yml.sample
+++ b/esb.yml.sample
@@ -30,6 +30,11 @@ services:
 flows:
   sample_flow:                          # The flow "code" and will be the Beanstalkd tube name
     description: Sample Flow            # The flow description
+    es_index:                           # Optional: the create/update ElasticSearch index API body (see https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-create-index.html). This is useful if you want to control index mapping, settings and aliases.
+        settings:                       # For example you can set the total_fields limit to an higher (or lower) value:
+            index:
+                total_fields:
+                    limit: 2000
     producer:
       service: My\Esb\Producer          # A producer service ID defined above
       batch_size: 1200                  # Jobs are produced in batches of 1200 jobs. Optional: default is 1000
diff --git a/src/FlowConfiguration.php b/src/FlowConfiguration.php
index 47ea2a6..5caa211 100644
--- a/src/FlowConfiguration.php
+++ b/src/FlowConfiguration.php
@@ -24,6 +24,7 @@ public function getConfigTreeBuilder(): TreeBuilder
             ->arrayPrototype()
                 ->children()
                     ->scalarNode('description')->isRequired()->cannotBeEmpty()->end()
+                    ->variableNode('es_index')->defaultNull()->end()
                     ->arrayNode('producer')
                         ->children()
                             ->scalarNode('service')->isRequired()->end()
diff --git a/src/Model/FlowConfig.php b/src/Model/FlowConfig.php
index 874174a..50f1066 100644
--- a/src/Model/FlowConfig.php
+++ b/src/Model/FlowConfig.php
@@ -102,4 +102,12 @@ public function getProducerBatchSize(): int
     {
         return $this->config['producer']['batch_size'];
     }
+
+    /**
+     * @return array<array-key, mixed>|null
+     */
+    public function getElasticSearchIndexCreateOrUpdateBody(): ?array
+    {
+        return $this->config['es_index'] ?? null;
+    }
 }
diff --git a/src/Service/ElasticSearch.php b/src/Service/ElasticSearch.php
index b614443..9d32ca2 100644
--- a/src/Service/ElasticSearch.php
+++ b/src/Service/ElasticSearch.php
@@ -91,6 +91,14 @@ public function getClient(): Client
         return $this->client;
     }
 
+    /**
+     * @param array<array-key, mixed>|null $createOrUpdateBody
+     */
+    public function setElasticSearchIndex(string $indexName, array $createOrUpdateBody = null): Amp\Promise
+    {
+        return $this->client->createOrUpdateIndex($indexName, $createOrUpdateBody);
+    }
+
     /**
      * @param JobInterface $job
      * @param string $indexName
diff --git a/src/Service/QueueManager.php b/src/Service/QueueManager.php
index f4206ec..ec6248f 100644
--- a/src/Service/QueueManager.php
+++ b/src/Service/QueueManager.php
@@ -72,6 +72,19 @@ public function __construct(
     public function boot(): Promise
     {
         return call(function () {
+            if ($this->flowConfig->getElasticSearchIndexCreateOrUpdateBody() !== null) {
+                yield $this->elasticSearch->setElasticSearchIndex(
+                    $this->flowConfig->getTube(),
+                    $this->flowConfig->getElasticSearchIndexCreateOrUpdateBody()
+                );
+                $this->logger->debug(
+                    'Successfully set ElasticSearch index',
+                    [
+                        'index' => $this->flowConfig->getTube(),
+                        'body' => $this->flowConfig->getElasticSearchIndexCreateOrUpdateBody()
+                    ]
+                );
+            }
             //Producer
             yield $this->beanstalkClient->use($this->flowConfig->getTube());