Merge pull request #1 from ensi-platform/task-83277

#83277

Author: arrilot

config/initial-event-propagation.php

@@ -5,6 +5,12 @@
     'set_initial_event_http_middleware' => [
         'default_user_type' => '',
 
+        /**
+         * If is set to `true` the middleware does not override the InitialEvent if it was already set for current context earlier.
+         * Defaults to `false`.
+         */
+        'preserve_existing_event' => false,
+
         /**
          * Middleware parses this header to get `appCode`.
          * If the header is not specified here or in a request, `appCode` is taken from `app_code` config value

src/RdKafkaConsumerMiddleware.php

@@ -0,0 +1,33 @@
+<?php
+
+namespace Ensi\LaravelInitialEventPropagation;
+
+use Closure;
+use Ensi\InitialEventPropagation\Config;
+use Ensi\InitialEventPropagation\InitialEventDTO;
+use Ensi\InitialEventPropagation\InitialEventHolder;
+use RdKafka\Message;
+
+class RdKafkaConsumerMiddleware
+{
+    public function __construct(private InitialEventHolder $initialEventHolder)
+    {
+    }
+
+    public function handle(Message $message, Closure $next): mixed
+    {
+        $initialEvent = $this->extractInitialEventFromHeaders($message->headers);
+        if ($initialEvent) {
+            $this->initialEventHolder->setInitialEvent($initialEvent);
+        }
+
+        return $next($message);
+    }
+
+    protected function extractInitialEventFromHeaders(?array $headers): ?InitialEventDTO
+    {
+        $header = $headers[Config::REQUEST_HEADER] ?? '';
+
+        return $header ? InitialEventDTO::fromSerializedString($header) : null;
+    }
+}

src/RdKafkaProducerMiddleware.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace Ensi\LaravelInitialEventPropagation;
+
+use Closure;
+use Ensi\InitialEventPropagation\Config;
+use Ensi\InitialEventPropagation\InitialEventHolder;
+use Ensi\LaravelPhpRdKafkaProducer\ProducerMessage;
+
+class RdKafkaProducerMiddleware
+{
+    public function __construct(private InitialEventHolder $initialEventHolder)
+    {
+    }
+
+    public function handle(ProducerMessage $message, Closure $next): mixed
+    {
+        $initialEvent = $this->initialEventHolder->getInitialEvent();
+        if ($initialEvent) {
+            $message->headers = $message->headers ?: [];
+            $message->headers[Config::REQUEST_HEADER] = $initialEvent->serialize();
+        }
+
+        return $next($message);
+    }
+}

src/SetInitialEventHttpMiddleware.php

@@ -15,17 +15,21 @@ public function handle(Request $request, Closure $next): mixed
         $user = $request->user();
         $config = config('initial-event-propagation', []);
         $mc = config('initial-event-propagation.set_initial_event_http_middleware', []);
-
-        $initialEvent = InitialEventDTO::fromScratch(
-            userId: $user ? $user->getId() : "",
-            userType: $user ? $mc['default_user_type'] : "",
-            app: !empty($mc['app_code_header']) ? $request->header($mc['app_code_header']) : ($config['app_code'] ?? ''),
-            entrypoint: $this->extractEntrypoint($request),
-            correlationId: !empty($mc['correlation_id_header']) ? $request->header($mc['correlation_id_header']) : '',
-            timestamp: !empty($mc['timestamp_header']) ? $request->header($mc['timestamp_header']) : ''
-        );
-
-        Container::getInstance()->make(InitialEventHolder::class)->setInitialEvent($initialEvent);
+        $holder =  Container::getInstance()->make(InitialEventHolder::class);
+        $existingEvent = $holder->getInitialEvent();
+
+        if (!$existingEvent || empty($mc['preserve_existing_event'])) {
+            $initialEvent = InitialEventDTO::fromScratch(
+                userId: $user ? $user->getId() : "",
+                userType: $user ? $mc['default_user_type'] : "",
+                app: !empty($mc['app_code_header']) ? $request->header($mc['app_code_header']) : ($config['app_code'] ?? ''),
+                entrypoint: $this->extractEntrypoint($request),
+                correlationId: !empty($mc['correlation_id_header']) ? $request->header($mc['correlation_id_header']) : '',
+                timestamp: !empty($mc['timestamp_header']) ? $request->header($mc['timestamp_header']) : ''
+            );
+
+            $holder->setInitialEvent($initialEvent);
+        }
 
         return $next($request);
     }