feat(dbal): replay dead letter messages to a chosen channel

packages/Dbal/Api/ExtensionObject/DbalDeadLetterBuilder.php

@@ -11,6 +11,7 @@
 use Ecotone\Messaging\Config\Container\MessagingContainerBuilder;
 use Ecotone\Messaging\Config\Container\Reference;
 use Ecotone\Messaging\Gateway\MessagingEntrypointService;
+use Ecotone\Messaging\Handler\ChannelResolver;
 use Ecotone\Messaging\Handler\InputOutputMessageHandlerBuilder;
 use Ecotone\Messaging\Handler\InterfaceToCall;
 use Ecotone\Messaging\Handler\InterfaceToCallRegistry;
@@ -27,6 +28,7 @@ class DbalDeadLetterBuilder extends InputOutputMessageHandlerBuilder
 {
     public const LIMIT_HEADER  = 'ecotone_dbal_deadletter_limit';
     public const OFFSET_HEADER = 'ecotone_dbal_deadletter_offset';
+    public const TARGET_CHANNEL_HEADER = 'ecotone_dbal_deadletter_target_channel';
 
     public const LIST_CHANNEL  = 'ecotone.dbal.deadletter.list';
     public const SHOW_CHANNEL       = 'ecotone.dbal.deadletter.show';
@@ -101,7 +103,11 @@ public static function createReplay(string $referenceName): self
             'replay',
             $referenceName,
             self::getChannelName($referenceName, self::REPLAY_CHANNEL),
-            []
+            [
+                PayloadBuilder::create('messageId'),
+                ReferenceBuilder::create('messagingEntrypoint', MessagingEntrypointService::class),
+                HeaderBuilder::createOptional('targetChannel', self::TARGET_CHANNEL_HEADER),
+            ]
         );
     }
 
@@ -113,6 +119,7 @@ public static function createReplayAll(string $referenceName): self
             self::getChannelName($referenceName, self::REPLAY_ALL_CHANNEL),
             [
                 ReferenceBuilder::create('messagingEntrypoint', MessagingEntrypointService::class),
+                HeaderBuilder::createOptional('targetChannel', self::TARGET_CHANNEL_HEADER),
             ]
         );
     }
@@ -160,6 +167,7 @@ public function compile(MessagingContainerBuilder $builder): Definition
                 Reference::to(self::storeReferenceFor($this->referenceName)),
                 Reference::to(MessageBuilder::class),
                 Reference::to(FlowTrackingScope::class),
+                Reference::to(ChannelResolver::class),
             ]));
         }
 

packages/Dbal/Api/Gateway/DeadLetterGateway.php

@@ -21,9 +21,9 @@ public function show(string $messageId): Message;
 
     public function count(): int;
 
-    public function replay(string|array $messageId): void;
+    public function replay(string|array $messageId, ?string $targetChannel = null): void;
 
-    public function replayAll(): void;
+    public function replayAll(?string $targetChannel = null): void;
 
     public function delete(string|array $messageId): void;
 

packages/Dbal/src/Recoverability/DbalDeadLetterConsoleCommand.php

@@ -21,8 +21,8 @@ public function help(): ConsoleCommandResultSet
             [
                 [DbalDeadLetterModule::LIST_COMMAND_NAME . ' {page:int}', 'List all dead messages'],
                 [DbalDeadLetterModule::SHOW_COMMAND_NAME . ' {messageId:string} {fullDetails:bool}', 'Show more details about specific message'],
-                [DbalDeadLetterModule::REPLAY_COMMAND_NAME . ' {messageId:string}', 'Replay chosen dead message'],
-                [DbalDeadLetterModule::REPLAY_ALL_COMMAND_NAME, 'Replay ALL dead messages'],
+                [DbalDeadLetterModule::REPLAY_COMMAND_NAME . ' {messageId:string} {targetChannel:?string}', 'Replay chosen dead message, to its source channel or to the target channel'],
+                [DbalDeadLetterModule::REPLAY_ALL_COMMAND_NAME . ' {targetChannel:?string}', 'Replay ALL dead messages, to their source channels or to the target channel'],
                 [DbalDeadLetterModule::DELETE_COMMAND_NAME . ' {messageId:string}', 'Delete chosen dead message'],
             ]
         );
@@ -65,19 +65,19 @@ public function show(DeadLetterGateway $deadLetterGateway, string $messageId, bo
         );
     }
 
-    public function replay(DeadLetterGateway $deadLetterGateway, string $messageId): void
+    public function replay(DeadLetterGateway $deadLetterGateway, string $messageId, ?string $targetChannel = null): void
     {
-        $deadLetterGateway->replay($messageId);
+        $deadLetterGateway->replay($messageId, $targetChannel);
     }
 
     public function delete(DeadLetterGateway $deadLetterGateway, string $messageId): void
     {
         $deadLetterGateway->delete($messageId);
     }
 
-    public function replayAll(DeadLetterGateway $deadLetterGateway): void
+    public function replayAll(DeadLetterGateway $deadLetterGateway, ?string $targetChannel = null): void
     {
-        $deadLetterGateway->replayAll();
+        $deadLetterGateway->replayAll($targetChannel);
     }
 
     private function convertTimestampToReadableFormat(int $timestamp)

packages/Dbal/src/Recoverability/DbalDeadLetterModule.php

@@ -210,6 +210,9 @@ private function registerGateway(string $referenceName, string $connectionFactor
                     'replay',
                     DbalDeadLetterBuilder::getChannelName($referenceName, DbalDeadLetterBuilder::REPLAY_CHANNEL)
                 )
+                    ->withParameterConverters([
+                        GatewayHeaderBuilder::create('targetChannel', DbalDeadLetterBuilder::TARGET_CHANNEL_HEADER),
+                    ])
             )
             ->registerGatewayBuilder(
                 GatewayProxyBuilder::create(
@@ -218,6 +221,9 @@ private function registerGateway(string $referenceName, string $connectionFactor
                     'replayAll',
                     DbalDeadLetterBuilder::getChannelName($referenceName, DbalDeadLetterBuilder::REPLAY_ALL_CHANNEL)
                 )
+                    ->withParameterConverters([
+                        GatewayHeaderBuilder::create('targetChannel', DbalDeadLetterBuilder::TARGET_CHANNEL_HEADER),
+                    ])
             )
             ->registerGatewayBuilder(
                 GatewayProxyBuilder::create(

packages/Dbal/src/Recoverability/DeadLetterHandler.php

@@ -11,6 +11,7 @@
 use Ecotone\Dbal\Tracing\FlowTrackingScope;
 use Ecotone\Messaging\Config\Annotation\ModuleConfiguration\ErrorChannelExceptionMessages;
 use Ecotone\Messaging\Gateway\MessagingEntrypointService;
+use Ecotone\Messaging\Handler\ChannelResolver;
 use Ecotone\Messaging\Support\InvalidArgumentException;
 use Ecotone\Messaging\Support\MessageBuilder;
 
@@ -23,6 +24,7 @@ public function __construct(
         private DeadLetterStore $deadLetterStore,
         private MessageBuilder $messageBuilder,
         private FlowTrackingScope $flowTracking,
+        private ChannelResolver $channelResolver,
     ) {
     }
 
@@ -60,18 +62,22 @@ public function count(): int
         return $this->deadLetterStore->count();
     }
 
-    public function replay(string|array $messageId, MessagingEntrypointService $messagingEntrypoint): void
+    public function replay(string|array $messageId, MessagingEntrypointService $messagingEntrypoint, ?string $targetChannel = null): void
     {
+        $this->assertTargetChannelExists($targetChannel);
+
         foreach ((array) $messageId as $id) {
-            $this->replayMessage($id, $messagingEntrypoint);
+            $this->replayMessage($id, $messagingEntrypoint, $targetChannel);
         }
     }
 
-    public function replayAll(MessagingEntrypointService $messagingEntrypoint): void
+    public function replayAll(MessagingEntrypointService $messagingEntrypoint, ?string $targetChannel = null): void
     {
+        $this->assertTargetChannelExists($targetChannel);
+
         while ($errorContexts = $this->deadLetterStore->list(100, 0)) {
             foreach ($errorContexts as $errorContext) {
-                $this->replayMessage($errorContext->getMessageId(), $messagingEntrypoint);
+                $this->replayMessage($errorContext->getMessageId(), $messagingEntrypoint, $targetChannel);
             }
         }
     }
@@ -93,18 +99,31 @@ public function store(Message $message): void
         $this->deadLetterStore->store($message);
     }
 
-    private function replayMessage(string $messageId, MessagingEntrypointService $messagingEntrypoint): void
+    private function assertTargetChannelExists(?string $targetChannel): void
+    {
+        if ($targetChannel !== null && ! $this->channelResolver->hasChannelWithName($targetChannel)) {
+            throw InvalidArgumentException::create(ErrorChannelExceptionMessages::cannotReplayToUnknownChannel($targetChannel));
+        }
+    }
+
+    private function replayMessage(string $messageId, MessagingEntrypointService $messagingEntrypoint, ?string $targetChannel): void
     {
         $message = $this->show($messageId);
         $hasPolledChannel = $message->getHeaders()->containsKey(MessageHeaders::POLLED_CHANNEL_NAME);
         $hasInboundRequestChannel = $message->getHeaders()->containsKey(MessageHeaders::INBOUND_REQUEST_CHANNEL);
         $hasRoutingSlip = $message->getHeaders()->containsKey(MessageHeaders::ROUTING_SLIP);
 
-        if (! $hasPolledChannel && ! $hasInboundRequestChannel && ! $hasRoutingSlip) {
+        if ($targetChannel === null && ! $hasPolledChannel && ! $hasInboundRequestChannel && ! $hasRoutingSlip) {
             throw InvalidArgumentException::create(ErrorChannelExceptionMessages::cannotReplyToDeadLetterMessage($messageId));
         }
 
-        if ($hasPolledChannel) {
+        if ($targetChannel !== null) {
+            $entrypoint = $targetChannel;
+
+            $message = $this->messageBuilder->fromMessage($message)
+                ->removeHeader(MessageHeaders::ROUTING_SLIP)
+                ->build();
+        } elseif ($hasPolledChannel) {
             $entrypoint = $message->getHeaders()->get(MessageHeaders::POLLED_CHANNEL_NAME);
         } elseif ($hasInboundRequestChannel) {
             $entrypoint = $message->getHeaders()->get(MessageHeaders::INBOUND_REQUEST_CHANNEL);