From bfa170620b6ac34f97f76e072991d64387d2b522 Mon Sep 17 00:00:00 2001
From: Brent Roose <brent.roose@gmail.com>
Date: Thu, 14 Nov 2024 13:57:22 +0100
Subject: [PATCH] feat(commandbus): async commands (#685)

---
 composer.json                                 |  1 +
 src/Tempest/CommandBus/src/AsyncCommand.php   | 12 +++
 .../CommandBus/src/AsyncCommandMiddleware.php | 38 ++++++++
 .../FileCommandRepository.php                 | 60 ++++++++++++
 .../MemoryRepository.php                      | 37 +++++++
 .../CommandBus/src/CommandBusConfig.php       |  4 +
 .../CommandBus/src/CommandRepository.php      | 19 ++++
 .../src/CommandRepositoryInitializer.php      | 20 ++++
 .../src/Exceptions/CouldNotResolveCommand.php | 11 +++
 .../CommandBus/src/HandleAsyncCommand.php     | 68 +++++++++++++
 .../CommandBus/src/MonitorAsyncCommands.php   | 95 ++++++++++++++++++
 .../CommandBus/src/stored-commands/.gitignore |  1 +
 .../src/Actions/ExecuteConsoleCommand.php     |  4 +-
 .../src/Exceptions/ConsoleErrorHandler.php    |  2 +-
 .../Console/src/Input/ConsoleArgumentBag.php  |  5 +
 .../Fixtures/Console/DispatchAsyncCommand.php | 32 ++++++
 .../Handlers/MyAsyncCommandHandler.php        | 27 ++++++
 .../CommandBus/AsyncCommandTest.php           | 97 +++++++++++++++++++
 .../CommandBus/Fixtures/MyAsyncCommand.php    | 16 +++
 .../Fixtures/MyFailingAsyncCommand.php        | 16 +++
 20 files changed, 562 insertions(+), 3 deletions(-)
 create mode 100644 src/Tempest/CommandBus/src/AsyncCommand.php
 create mode 100644 src/Tempest/CommandBus/src/AsyncCommandMiddleware.php
 create mode 100644 src/Tempest/CommandBus/src/AsyncCommandRepositories/FileCommandRepository.php
 create mode 100644 src/Tempest/CommandBus/src/AsyncCommandRepositories/MemoryRepository.php
 create mode 100644 src/Tempest/CommandBus/src/CommandRepository.php
 create mode 100644 src/Tempest/CommandBus/src/CommandRepositoryInitializer.php
 create mode 100644 src/Tempest/CommandBus/src/Exceptions/CouldNotResolveCommand.php
 create mode 100644 src/Tempest/CommandBus/src/HandleAsyncCommand.php
 create mode 100644 src/Tempest/CommandBus/src/MonitorAsyncCommands.php
 create mode 100644 src/Tempest/CommandBus/src/stored-commands/.gitignore
 create mode 100644 tests/Fixtures/Console/DispatchAsyncCommand.php
 create mode 100644 tests/Fixtures/Handlers/MyAsyncCommandHandler.php
 create mode 100644 tests/Integration/CommandBus/AsyncCommandTest.php
 create mode 100644 tests/Integration/CommandBus/Fixtures/MyAsyncCommand.php
 create mode 100644 tests/Integration/CommandBus/Fixtures/MyFailingAsyncCommand.php

diff --git a/composer.json b/composer.json
index 774974658..13625164a 100644
--- a/composer.json
+++ b/composer.json
@@ -30,6 +30,7 @@
         "psr/http-message": "^1.0|^2.0",
         "psr/log": "^3.0.0",
         "symfony/cache": "^7.2",
+        "symfony/process": "^7.1",
         "symfony/uid": "^7.1",
         "symfony/var-dumper": "^7.1",
         "symfony/var-exporter": "^7.1",
diff --git a/src/Tempest/CommandBus/src/AsyncCommand.php b/src/Tempest/CommandBus/src/AsyncCommand.php
new file mode 100644
index 000000000..a3af36271
--- /dev/null
+++ b/src/Tempest/CommandBus/src/AsyncCommand.php
@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tempest\CommandBus;
+
+use Attribute;
+
+#[Attribute]
+final readonly class AsyncCommand
+{
+}
diff --git a/src/Tempest/CommandBus/src/AsyncCommandMiddleware.php b/src/Tempest/CommandBus/src/AsyncCommandMiddleware.php
new file mode 100644
index 000000000..b79ef2a30
--- /dev/null
+++ b/src/Tempest/CommandBus/src/AsyncCommandMiddleware.php
@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tempest\CommandBus;
+
+use Symfony\Component\Uid\Uuid;
+use Tempest\Core\KernelEvent;
+use Tempest\EventBus\EventHandler;
+use Tempest\Reflection\ClassReflector;
+
+final readonly class AsyncCommandMiddleware implements CommandBusMiddleware
+{
+    public function __construct(
+        private CommandBusConfig $commandBusConfig,
+        private CommandRepository $repository,
+    ) {
+    }
+
+    #[EventHandler(KernelEvent::BOOTED)]
+    public function onBooted(): void
+    {
+        $this->commandBusConfig->addMiddleware(self::class);
+    }
+
+    public function __invoke(object $command, CommandBusMiddlewareCallable $next): void
+    {
+        $reflector = new ClassReflector($command);
+
+        if ($reflector->hasAttribute(AsyncCommand::class)) {
+            $this->repository->store(Uuid::v7()->toString(), $command);
+
+            return;
+        }
+
+        $next($command);
+    }
+}
diff --git a/src/Tempest/CommandBus/src/AsyncCommandRepositories/FileCommandRepository.php b/src/Tempest/CommandBus/src/AsyncCommandRepositories/FileCommandRepository.php
new file mode 100644
index 000000000..3c6a2cb04
--- /dev/null
+++ b/src/Tempest/CommandBus/src/AsyncCommandRepositories/FileCommandRepository.php
@@ -0,0 +1,60 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tempest\CommandBus\AsyncCommandRepositories;
+
+use Tempest\CommandBus\CommandRepository;
+use Tempest\CommandBus\Exceptions\CouldNotResolveCommand;
+use function Tempest\Support\arr;
+
+final readonly class FileCommandRepository implements CommandRepository
+{
+    public function store(string $uuid, object $command): void
+    {
+        $payload = serialize($command);
+
+        file_put_contents(__DIR__ . "/../stored-commands/{$uuid}.pending.txt", $payload);
+    }
+
+    public function findPendingCommand(string $uuid): object
+    {
+        $path = __DIR__ . "/../stored-commands/{$uuid}.pending.txt";
+
+        if (! file_exists($path)) {
+            throw new CouldNotResolveCommand($uuid);
+        }
+
+        $payload = file_get_contents($path);
+
+        return unserialize($payload);
+    }
+
+    public function markAsDone(string $uuid): void
+    {
+        $path = __DIR__ . "/../stored-commands/{$uuid}.pending.txt";
+
+        unlink($path);
+    }
+
+    public function markAsFailed(string $uuid): void
+    {
+        rename(
+            from: __DIR__ . "/../stored-commands/{$uuid}.pending.txt",
+            to: __DIR__ . "/../stored-commands/{$uuid}.failed.txt",
+        );
+    }
+
+    public function getPendingCommands(): array
+    {
+        return arr(glob(__DIR__ . "/../stored-commands/*.pending.txt"))
+            ->mapWithKeys(function (string $path) {
+                $uuid = str_replace('.pending.txt', '', pathinfo($path, PATHINFO_BASENAME));
+
+                $payload = file_get_contents($path);
+
+                yield $uuid => unserialize($payload);
+            })
+            ->toArray();
+    }
+}
diff --git a/src/Tempest/CommandBus/src/AsyncCommandRepositories/MemoryRepository.php b/src/Tempest/CommandBus/src/AsyncCommandRepositories/MemoryRepository.php
new file mode 100644
index 000000000..23a178d73
--- /dev/null
+++ b/src/Tempest/CommandBus/src/AsyncCommandRepositories/MemoryRepository.php
@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tempest\CommandBus\AsyncCommandRepositories;
+
+use Tempest\CommandBus\CommandRepository;
+
+final class MemoryRepository implements CommandRepository
+{
+    private array $commands = [];
+
+    public function store(string $uuid, object $command): void
+    {
+        $this->commands[$uuid] = $command;
+    }
+
+    public function getPendingCommands(): array
+    {
+        return $this->commands;
+    }
+
+    public function findPendingCommand(string $uuid): object
+    {
+        return $this->commands[$uuid];
+    }
+
+    public function markAsDone(string $uuid): void
+    {
+        unset($this->commands[$uuid]);
+    }
+
+    public function markAsFailed(string $uuid): void
+    {
+        unset($this->commands[$uuid]);
+    }
+}
diff --git a/src/Tempest/CommandBus/src/CommandBusConfig.php b/src/Tempest/CommandBus/src/CommandBusConfig.php
index a678cee48..4891d11dc 100644
--- a/src/Tempest/CommandBus/src/CommandBusConfig.php
+++ b/src/Tempest/CommandBus/src/CommandBusConfig.php
@@ -4,6 +4,7 @@
 
 namespace Tempest\CommandBus;
 
+use Tempest\CommandBus\AsyncCommandRepositories\FileCommandRepository;
 use Tempest\Reflection\MethodReflector;
 
 final class CommandBusConfig
@@ -14,6 +15,9 @@ public function __construct(
 
         /** @var array<array-key, class-string<\Tempest\CommandBus\CommandBusMiddleware>> */
         public array $middleware = [],
+
+        /** @var class-string<\Tempest\CommandBus\CommandRepository> $commandRepositoryClass */
+        public string $commandRepositoryClass = FileCommandRepository::class,
     ) {
     }
 
diff --git a/src/Tempest/CommandBus/src/CommandRepository.php b/src/Tempest/CommandBus/src/CommandRepository.php
new file mode 100644
index 000000000..1a0b35d1b
--- /dev/null
+++ b/src/Tempest/CommandBus/src/CommandRepository.php
@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tempest\CommandBus;
+
+interface CommandRepository
+{
+    public function store(string $uuid, object $command): void;
+
+    /** @return array<string, object> */
+    public function getPendingCommands(): array;
+
+    public function findPendingCommand(string $uuid): object;
+
+    public function markAsDone(string $uuid): void;
+
+    public function markAsFailed(string $uuid): void;
+}
diff --git a/src/Tempest/CommandBus/src/CommandRepositoryInitializer.php b/src/Tempest/CommandBus/src/CommandRepositoryInitializer.php
new file mode 100644
index 000000000..b2afa16a2
--- /dev/null
+++ b/src/Tempest/CommandBus/src/CommandRepositoryInitializer.php
@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tempest\CommandBus;
+
+use Tempest\Container\Container;
+use Tempest\Container\Initializer;
+use Tempest\Container\Singleton;
+
+final readonly class CommandRepositoryInitializer implements Initializer
+{
+    #[Singleton]
+    public function initialize(Container $container): CommandRepository
+    {
+        $commandRepositoryClass = $container->get(CommandBusConfig::class)->commandRepositoryClass;
+
+        return $container->get($commandRepositoryClass);
+    }
+}
diff --git a/src/Tempest/CommandBus/src/Exceptions/CouldNotResolveCommand.php b/src/Tempest/CommandBus/src/Exceptions/CouldNotResolveCommand.php
new file mode 100644
index 000000000..483b836e0
--- /dev/null
+++ b/src/Tempest/CommandBus/src/Exceptions/CouldNotResolveCommand.php
@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tempest\CommandBus\Exceptions;
+
+use Exception;
+
+final class CouldNotResolveCommand extends Exception
+{
+}
diff --git a/src/Tempest/CommandBus/src/HandleAsyncCommand.php b/src/Tempest/CommandBus/src/HandleAsyncCommand.php
new file mode 100644
index 000000000..79ede0702
--- /dev/null
+++ b/src/Tempest/CommandBus/src/HandleAsyncCommand.php
@@ -0,0 +1,68 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tempest\CommandBus;
+
+use Tempest\Console\Console;
+use Tempest\Console\ConsoleCommand;
+use Tempest\Console\ExitCode;
+use Tempest\Console\HasConsole;
+use Tempest\Container\Container;
+use function Tempest\Support\arr;
+use Throwable;
+
+final readonly class HandleAsyncCommand
+{
+    use HasConsole;
+
+    public function __construct(
+        private CommandBusConfig $commandBusConfig,
+        private Container $container,
+        private Console $console,
+        private CommandRepository $repository,
+    ) {
+    }
+
+    #[ConsoleCommand(name: 'command:handle')]
+    public function __invoke(?string $uuid = null): ExitCode
+    {
+        try {
+            if ($uuid) {
+                $command = $this->repository->findPendingCommand($uuid);
+            } else {
+                $command = arr($this->repository->getPendingCommands())->first();
+            }
+
+            if (! $command) {
+                $this->error('No pending command found');
+
+                return ExitCode::ERROR;
+            }
+
+            $commandHandler = $this->commandBusConfig->handlers[$command::class] ?? null;
+
+            if (! $commandHandler) {
+                $commandClass = $command::class;
+                $this->error("No handler found for command {$commandClass}");
+
+                return ExitCode::ERROR;
+            }
+
+            $commandHandler->handler->invokeArgs(
+                $this->container->get($commandHandler->handler->getDeclaringClass()->getName()),
+                [$command],
+            );
+
+            $this->repository->markAsDone($uuid);
+            $this->success('Done');
+
+            return ExitCode::SUCCESS;
+        } catch (Throwable $throwable) {
+            $this->repository->markAsFailed($uuid);
+            $this->error($throwable->getMessage());
+
+            return ExitCode::ERROR;
+        }
+    }
+}
diff --git a/src/Tempest/CommandBus/src/MonitorAsyncCommands.php b/src/Tempest/CommandBus/src/MonitorAsyncCommands.php
new file mode 100644
index 000000000..3508ab027
--- /dev/null
+++ b/src/Tempest/CommandBus/src/MonitorAsyncCommands.php
@@ -0,0 +1,95 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tempest\CommandBus;
+
+use DateTimeImmutable;
+use Symfony\Component\Process\Process;
+use Tempest\Console\Console;
+use Tempest\Console\ConsoleCommand;
+use Tempest\Console\HasConsole;
+use Tempest\Console\Input\ConsoleArgumentBag;
+use function Tempest\Support\arr;
+
+final readonly class MonitorAsyncCommands
+{
+    use HasConsole;
+
+    public function __construct(
+        private CommandRepository $repository,
+        private ConsoleArgumentBag $argumentBag,
+        private Console $console,
+    ) {
+    }
+
+    #[ConsoleCommand(name: 'command:monitor')]
+    public function __invoke(): void
+    {
+        $this->success("Monitoring for new commands. Press ctrl+c to stop.");
+
+        /** @var \Symfony\Component\Process\Process[] $processes */
+        $processes = [];
+
+        while (true) { // @phpstan-ignore-line
+            foreach ($processes as $uuid => $process) {
+                $time = new DateTimeImmutable();
+
+                if ($process->isTerminated()) {
+                    if ($process->isSuccessful()) {
+                        $this->writeln("<success>{$uuid}</success> finished at {$time->format('Y-m-d H:i:s')}");
+                    } else {
+                        $this->writeln("<error>{$uuid}</error> failed at {$time->format('Y-m-d H:i:s')}");
+                    }
+
+                    if ($output = trim($process->getOutput())) {
+                        $this->writeln($output);
+                    }
+
+                    if ($errorOutput = trim($process->getErrorOutput())) {
+                        $this->writeln($errorOutput);
+                    }
+
+                    unset($processes[$uuid]);
+                }
+            }
+
+            $availableCommands = arr($this->repository->getPendingCommands())
+                ->filter(fn (object $command, string $uuid) => ! in_array($uuid, array_keys($processes)));
+
+            if (count($processes) === 5) {
+                $this->sleep(0.5);
+
+                continue;
+            }
+
+            if ($availableCommands->isEmpty()) {
+                $this->sleep(0.5);
+
+                continue;
+            }
+
+            // Start a task
+            $uuid = $availableCommands->keys()->first();
+
+            $time = new DateTimeImmutable();
+            $this->writeln("<h2>{$uuid}</h2> started at {$time->format('Y-m-d H:i:s')}");
+
+            $process = new Process([
+                $this->argumentBag->getBinaryPath(),
+                $this->argumentBag->getCliName(),
+                'command:handle',
+                $uuid,
+            ], getcwd());
+
+            $process->start();
+
+            $processes[$uuid] = $process;
+        }
+    }
+
+    private function sleep(float $seconds): void
+    {
+        usleep((int) ($seconds * 1_000_000));
+    }
+}
diff --git a/src/Tempest/CommandBus/src/stored-commands/.gitignore b/src/Tempest/CommandBus/src/stored-commands/.gitignore
new file mode 100644
index 000000000..314f02b1b
--- /dev/null
+++ b/src/Tempest/CommandBus/src/stored-commands/.gitignore
@@ -0,0 +1 @@
+*.txt
\ No newline at end of file
diff --git a/src/Tempest/Console/src/Actions/ExecuteConsoleCommand.php b/src/Tempest/Console/src/Actions/ExecuteConsoleCommand.php
index 09f849b3c..a799c3105 100644
--- a/src/Tempest/Console/src/Actions/ExecuteConsoleCommand.php
+++ b/src/Tempest/Console/src/Actions/ExecuteConsoleCommand.php
@@ -43,12 +43,12 @@ private function getCallable(array $commandMiddleware): ConsoleMiddlewareCallabl
 
             $inputBuilder = new ConsoleInputBuilder($consoleCommand, $invocation->argumentBag);
 
-            $consoleCommand->handler->invokeArgs(
+            $exitCode = $consoleCommand->handler->invokeArgs(
                 $consoleCommandClass,
                 $inputBuilder->build(),
             );
 
-            return ExitCode::SUCCESS;
+            return $exitCode ?? ExitCode::SUCCESS;
         });
 
         $middlewareStack = [...$this->consoleConfig->middleware, ...$commandMiddleware];
diff --git a/src/Tempest/Console/src/Exceptions/ConsoleErrorHandler.php b/src/Tempest/Console/src/Exceptions/ConsoleErrorHandler.php
index 0fbd29eef..dab363c1b 100644
--- a/src/Tempest/Console/src/Exceptions/ConsoleErrorHandler.php
+++ b/src/Tempest/Console/src/Exceptions/ConsoleErrorHandler.php
@@ -57,7 +57,7 @@ public function handleException(Throwable $throwable): void
 
     public function handleError(int $errNo, string $errstr, string $errFile, int $errLine): void
     {
-        ll('error');
+        ll(error: $errstr);
 
         $this->console
             ->writeln()
diff --git a/src/Tempest/Console/src/Input/ConsoleArgumentBag.php b/src/Tempest/Console/src/Input/ConsoleArgumentBag.php
index 73764aa2a..a29abde15 100644
--- a/src/Tempest/Console/src/Input/ConsoleArgumentBag.php
+++ b/src/Tempest/Console/src/Input/ConsoleArgumentBag.php
@@ -159,6 +159,11 @@ public function add(ConsoleInputArgument $argument): self
         return $this;
     }
 
+    public function getBinaryPath(): string
+    {
+        return PHP_BINARY;
+    }
+
     public function getCliName(): string
     {
         return $this->path[0] ?? '';
diff --git a/tests/Fixtures/Console/DispatchAsyncCommand.php b/tests/Fixtures/Console/DispatchAsyncCommand.php
new file mode 100644
index 000000000..248313800
--- /dev/null
+++ b/tests/Fixtures/Console/DispatchAsyncCommand.php
@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Tempest\Fixtures\Console;
+
+use function Tempest\command;
+use Tempest\Console\ConsoleCommand;
+use Tempest\Console\HasConsole;
+use Tests\Tempest\Integration\CommandBus\Fixtures\MyAsyncCommand;
+use Tests\Tempest\Integration\CommandBus\Fixtures\MyFailingAsyncCommand;
+
+final readonly class DispatchAsyncCommand
+{
+    use HasConsole;
+
+    #[ConsoleCommand(name: 'command:dispatch')]
+    public function __invoke(int $times = 10, bool $fail = false): void
+    {
+        foreach (range(1, $times) as $i) {
+            $command = $fail
+                ? new MyFailingAsyncCommand("{$i}")
+                : new MyAsyncCommand("{$i}");
+
+            command(
+                $command,
+            );
+        }
+
+        $this->info('Dispatched commands');
+    }
+}
diff --git a/tests/Fixtures/Handlers/MyAsyncCommandHandler.php b/tests/Fixtures/Handlers/MyAsyncCommandHandler.php
new file mode 100644
index 000000000..b9466e14d
--- /dev/null
+++ b/tests/Fixtures/Handlers/MyAsyncCommandHandler.php
@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Tempest\Fixtures\Handlers;
+
+use Exception;
+use Tempest\CommandBus\CommandHandler;
+use Tests\Tempest\Integration\CommandBus\Fixtures\MyAsyncCommand;
+use Tests\Tempest\Integration\CommandBus\Fixtures\MyFailingAsyncCommand;
+
+final class MyAsyncCommandHandler
+{
+    public static bool $isHandled = false;
+
+    #[CommandHandler]
+    public function onMyAsyncCommand(MyAsyncCommand $command): void
+    {
+        self::$isHandled = true;
+    }
+
+    #[CommandHandler]
+    public function onMyFailingAsyncCommand(MyFailingAsyncCommand $command): void
+    {
+        throw new Exception('Failed command');
+    }
+}
diff --git a/tests/Integration/CommandBus/AsyncCommandTest.php b/tests/Integration/CommandBus/AsyncCommandTest.php
new file mode 100644
index 000000000..da40850de
--- /dev/null
+++ b/tests/Integration/CommandBus/AsyncCommandTest.php
@@ -0,0 +1,97 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Tempest\Integration\CommandBus;
+
+use Symfony\Component\Process\Process;
+use function Tempest\command;
+use Tempest\CommandBus\AsyncCommandRepositories\MemoryRepository;
+use Tempest\CommandBus\CommandRepository;
+use Tempest\Highlight\Themes\TerminalStyle;
+use function Tempest\Support\arr;
+use Tests\Tempest\Fixtures\Handlers\MyAsyncCommandHandler;
+use Tests\Tempest\Integration\CommandBus\Fixtures\MyAsyncCommand;
+use Tests\Tempest\Integration\FrameworkIntegrationTestCase;
+
+/**
+ * @internal
+ */
+final class AsyncCommandTest extends FrameworkIntegrationTestCase
+{
+    public function test_async_commands_are_stored_and_handled_afterwards(): void
+    {
+        $repository = new MemoryRepository();
+
+        $this->container->singleton(
+            CommandRepository::class,
+            fn () => $repository,
+        );
+
+        MyAsyncCommandHandler::$isHandled = false;
+
+        command(new MyAsyncCommand('Brent'));
+
+        $pendingCommands = arr($repository->getPendingCommands());
+
+        $this->assertCount(1, $pendingCommands);
+        $command = $pendingCommands->first();
+        $this->assertSame('Brent', $command->name);
+        $this->assertFalse(MyAsyncCommandHandler::$isHandled);
+
+        $this->console
+            ->call("command:handle " . $pendingCommands->keys()->first())
+            ->assertSee('Done');
+
+        $this->assertTrue(MyAsyncCommandHandler::$isHandled);
+    }
+
+    public function test_async_command_monitor(): void
+    {
+        $process = new Process(['php', 'tempest', 'command:monitor']);
+        $process->start();
+
+        $this->console->call("command:dispatch 1");
+
+        sleep(1);
+
+        $output = $this->getOutput($process);
+        $this->assertStringContainsString('Monitoring for new commands', $output);
+        $this->assertStringContainsString('started at', $output);
+        $this->assertStringContainsString('finished at', $output);
+        $this->assertStringContainsString('Done', $output);
+        $process->stop();
+    }
+
+    public function test_async_failed_command_monitor(): void
+    {
+        $process = new Process(['php', 'tempest', 'command:monitor']);
+        $process->start();
+
+        $this->console->call("command:dispatch 1 --fail");
+
+        sleep(1);
+
+        $output = $this->getOutput($process);
+        $this->assertStringContainsString('Monitoring for new commands', $output);
+        $this->assertStringContainsString('started at', $output);
+        $this->assertStringContainsString('failed at', $output);
+        $this->assertStringContainsString('Failed command', $output);
+        $process->stop();
+
+        arr(glob(__DIR__ . '/../../../src/Tempest/CommandBus/src/stored-commands/*.failed.txt'))
+            ->each(function (string $filename): void {
+                unlink($filename);
+            });
+    }
+
+    private function getOutput(Process $process): string
+    {
+        $pattern = array_map(
+            fn (TerminalStyle $consoleStyle) => TerminalStyle::ESC->value . $consoleStyle->value,
+            TerminalStyle::cases(),
+        );
+
+        return str_replace($pattern, '', $process->getOutput());
+    }
+}
diff --git a/tests/Integration/CommandBus/Fixtures/MyAsyncCommand.php b/tests/Integration/CommandBus/Fixtures/MyAsyncCommand.php
new file mode 100644
index 000000000..02bb32895
--- /dev/null
+++ b/tests/Integration/CommandBus/Fixtures/MyAsyncCommand.php
@@ -0,0 +1,16 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Tempest\Integration\CommandBus\Fixtures;
+
+use Tempest\CommandBus\AsyncCommand;
+
+#[AsyncCommand]
+final readonly class MyAsyncCommand
+{
+    public function __construct(
+        public string $name,
+    ) {
+    }
+}
diff --git a/tests/Integration/CommandBus/Fixtures/MyFailingAsyncCommand.php b/tests/Integration/CommandBus/Fixtures/MyFailingAsyncCommand.php
new file mode 100644
index 000000000..bb65aca41
--- /dev/null
+++ b/tests/Integration/CommandBus/Fixtures/MyFailingAsyncCommand.php
@@ -0,0 +1,16 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Tempest\Integration\CommandBus\Fixtures;
+
+use Tempest\CommandBus\AsyncCommand;
+
+#[AsyncCommand]
+final readonly class MyFailingAsyncCommand
+{
+    public function __construct(
+        public string $name,
+    ) {
+    }
+}