diff --git a/.github/workflows/unit.yml b/.github/workflows/unit.yml index 269c23682..880a35d8c 100644 --- a/.github/workflows/unit.yml +++ b/.github/workflows/unit.yml @@ -48,6 +48,11 @@ jobs: ini-values: memory_limit=-1 extensions: pdo_sqlite, pdo_pgsql + # symfony/tui needs PHP 8.4 and would raise the lowest symfony versions + - name: "Remove symfony/tui" + if: matrix.php-version == '8.2' || matrix.php-version == '8.3' || matrix.dependencies == 'lowest' + run: composer remove --dev symfony/tui --no-update + - uses: ramsey/composer-install@4.0.0 with: dependency-versions: ${{ matrix.dependencies }} diff --git a/bin/subscription-dashboard-demo b/bin/subscription-dashboard-demo new file mode 100755 index 000000000..475ef89ef --- /dev/null +++ b/bin/subscription-dashboard-demo @@ -0,0 +1,229 @@ +#!/usr/bin/env php +. + * + * bin/subscription-dashboard-demo [--no-worker] [--refresh=1] + */ + +declare(strict_types=1); + +namespace Patchlevel\EventSourcing\Demo; + +use Patchlevel\EventSourcing\Attribute\Processor; +use Patchlevel\EventSourcing\Attribute\Projector; +use Patchlevel\EventSourcing\Attribute\Subscribe; +use Patchlevel\EventSourcing\Clock\SystemClock; +use Patchlevel\EventSourcing\Console\Command\SubscriptionDashboardCommand; +use Patchlevel\EventSourcing\Message\Message; +use Patchlevel\EventSourcing\Store\Header\IndexHeader; +use Patchlevel\EventSourcing\Store\InMemoryStore; +use Patchlevel\EventSourcing\Subscription\Engine\Command\Boot; +use Patchlevel\EventSourcing\Subscription\Engine\Command\Pause; +use Patchlevel\EventSourcing\Subscription\Engine\Command\Run; +use Patchlevel\EventSourcing\Subscription\Engine\Command\Setup; +use Patchlevel\EventSourcing\Subscription\Engine\DefaultSubscriptionEngine; +use Patchlevel\EventSourcing\Subscription\Engine\StoreMessageLoader; +use Patchlevel\EventSourcing\Subscription\RetryStrategy\ClockBasedRetryStrategy; +use Patchlevel\EventSourcing\Subscription\RetryStrategy\RetryStrategyRepository; +use Patchlevel\EventSourcing\Subscription\RunMode; +use Patchlevel\EventSourcing\Subscription\Status; +use Patchlevel\EventSourcing\Subscription\Store\InMemorySubscriptionStore; +use Patchlevel\EventSourcing\Subscription\Subscriber\MetadataSubscriberAccessorRepository; +use Patchlevel\EventSourcing\Subscription\Subscription; +use Revolt\EventLoop; +use RuntimeException; +use Symfony\Component\Console\Application; +use Symfony\Component\Console\Input\ArgvInput; +use Throwable; + +use function array_filter; +use function array_values; +use function in_array; +use function random_int; +use function sprintf; +use function usleep; + +require __DIR__ . '/../vendor/autoload.php'; + +final class ProfileCreated +{ + public function __construct(public readonly string $name) + { + } +} + +final class ProfileVisited +{ + public function __construct(public readonly string $profile) + { + } +} + +final class OrderPlaced +{ + public function __construct(public readonly int $amount) + { + } +} + +#[Projector('profile')] +final class ProfileProjector +{ + #[Subscribe(ProfileCreated::class)] + public function onProfileCreated(ProfileCreated $event): void + { + } +} + +#[Projector('guest_list')] +final class GuestListProjector +{ + #[Subscribe(ProfileCreated::class)] + #[Subscribe(ProfileVisited::class)] + public function handle(Message $message): void + { + } +} + +/** A slow projector that needs a while to boot. */ +#[Projector('hotel_overview')] +final class HotelOverviewProjector +{ + #[Subscribe(Subscribe::ALL)] + public function handle(Message $message): void + { + usleep(2_000); + } +} + +/** Fails from time to time, so it switches between error, retries and failed. */ +#[Projector('profile_stats')] +final class ProfileStatsProjector +{ + #[Subscribe(ProfileVisited::class)] + public function onProfileVisited(Message $message): void + { + $index = $message->header(IndexHeader::class)->index; + + if ($index % 150 === 0 && random_int(1, 10) <= 7) { + throw new RuntimeException(sprintf( + 'SQLSTATE[40001]: Serialization failure: 1213 Deadlock found when trying to get lock (index %d)', + $index, + )); + } + } +} + +#[Projector('invoice_search')] +final class InvoiceSearchProjector +{ + #[Subscribe(OrderPlaced::class)] + public function onOrderPlaced(OrderPlaced $event): void + { + } +} + +#[Projector('legacy_report', runMode: RunMode::Once)] +final class LegacyReportProjector +{ + #[Subscribe(OrderPlaced::class)] + public function onOrderPlaced(OrderPlaced $event): void + { + } +} + +#[Projector('room_availability')] +final class RoomAvailabilityProjector +{ + #[Subscribe(OrderPlaced::class)] + public function onOrderPlaced(OrderPlaced $event): void + { + } +} + +#[Processor('send_welcome_mail')] +final class SendWelcomeMailProcessor +{ + #[Subscribe(ProfileCreated::class)] + public function onProfileCreated(ProfileCreated $event): void + { + } +} + +$input = new ArgvInput(); +$worker = !$input->hasParameterOption('--no-worker'); + +$store = new InMemoryStore(); + +$produce = static function (int $count) use ($store): void { + $messages = []; + + for ($i = 0; $i < $count; $i++) { + $messages[] = Message::create(match (random_int(1, 3)) { + 1 => new ProfileCreated('guest' . random_int(1, 10_000)), + 2 => new ProfileVisited('guest' . random_int(1, 10_000)), + default => new OrderPlaced(random_int(10, 500)), + }); + } + + $store->save(...$messages); +}; + +$produce(2_500); + +$engine = new DefaultSubscriptionEngine( + new StoreMessageLoader($store), + new InMemorySubscriptionStore([ + // a subscription without subscriber, it will be detached + new Subscription('old_newsletter', 'processor', RunMode::FromNow, Status::Active, 1_200), + ]), + new MetadataSubscriberAccessorRepository([ + new ProfileProjector(), + new GuestListProjector(), + new HotelOverviewProjector(), + new ProfileStatsProjector(), + new InvoiceSearchProjector(), + new LegacyReportProjector(), + new RoomAvailabilityProjector(), + new SendWelcomeMailProcessor(), + ]), + new RetryStrategyRepository([ + RetryStrategyRepository::DEFAULT_STRATEGY_NAME => new ClockBasedRetryStrategy(new SystemClock(), 3, 1.5, 3), + ]), +); + +$all = ['profile', 'guest_list', 'hotel_overview', 'profile_stats', 'invoice_search', 'legacy_report', 'send_welcome_mail', 'old_newsletter']; + +// room_availability stays "new", so it can be set up from the dashboard +$engine->execute(new Setup($all, skipBooting: true)); +$engine->execute(new Boot(['profile', 'guest_list', 'profile_stats', 'invoice_search', 'legacy_report', 'send_welcome_mail'])); +$engine->execute(new Pause(['invoice_search'])); + +EventLoop::repeat(0.5, static function () use ($produce): void { + $produce(random_int(0, 8)); +}); + +if ($worker) { + EventLoop::repeat(1.0, static function () use ($engine): void { + try { + $engine->execute(new Boot(null, null, 100)); + $engine->execute(new Run(null, null, 200)); + } catch (Throwable) { + // the dashboard shows the errors + } + }); +} + +$application = new Application('subscription dashboard demo'); +$application->addCommand(new SubscriptionDashboardCommand($engine, $store)); +$application->setDefaultCommand('event-sourcing:subscription:dashboard', true); + +$application->run(new ArgvInput(array_values(array_filter( + $_SERVER['argv'], + static fn (string $arg) => !in_array($arg, ['--no-worker'], true), +)))); diff --git a/composer.json b/composer.json index 0961bd096..881d731bc 100644 --- a/composer.json +++ b/composer.json @@ -61,11 +61,13 @@ "phpunit/phpunit": "^11.5.15", "symfony/cache": "^6.4.0 || ^7.0.0 || ^8.0.0", "symfony/messenger": "^5.4.31 || ^6.4.0 || ^7.0.1 || ^8.0.0", + "symfony/tui": "^8.1.0", "symfony/var-dumper": "^5.4.29 || ^6.4.0 || ^7.0.0 || ^8.0.0", "wnx/commonmark-markdown-renderer": "^1.5.0" }, "suggest": { - "patchlevel/event-sourcing-psalm-plugin": "for psalm support" + "patchlevel/event-sourcing-psalm-plugin": "for psalm support", + "symfony/tui": "for the interactive subscription dashboard (requires PHP 8.4+)" }, "config": { "preferred-install": { diff --git a/composer.lock b/composer.lock index d31f55436..7d32fbc2d 100644 --- a/composer.lock +++ b/composer.lock @@ -4,7 +4,7 @@ "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies", "This file is @generated automatically" ], - "content-hash": "c703d9a37598197f2f8bceee5da066c0", + "content-hash": "9f7d8dc2d4246f9ff31777368a560475", "packages": [ { "name": "brick/math", @@ -5484,6 +5484,78 @@ ], "time": "2026-02-18T12:37:06+00:00" }, + { + "name": "revolt/event-loop", + "version": "v1.0.9", + "source": { + "type": "git", + "url": "https://github.com/revoltphp/event-loop.git", + "reference": "44061cf513e53c6200372fc935ac42271566295d" + }, + "dist": { + "type": "zip", + "url": "https://api.github.com/repos/revoltphp/event-loop/zipball/44061cf513e53c6200372fc935ac42271566295d", + "reference": "44061cf513e53c6200372fc935ac42271566295d", + "shasum": "" + }, + "require": { + "php": ">=8.1" + }, + "require-dev": { + "ext-json": "*", + "jetbrains/phpstorm-stubs": "^2019.3", + "phpunit/phpunit": "^9", + "psalm/phar": "6.16.*" + }, + "type": "library", + "extra": { + "branch-alias": { + "dev-main": "1.x-dev" + } + }, + "autoload": { + "psr-4": { + "Revolt\\": "src" + } + }, + "notification-url": "https://packagist.org/downloads/", + "license": [ + "MIT" + ], + "authors": [ + { + "name": "Aaron Piotrowski", + "email": "aaron@trowski.com" + }, + { + "name": "Cees-Jan Kiewiet", + "email": "ceesjank@gmail.com" + }, + { + "name": "Christian Lück", + "email": "christian@clue.engineering" + }, + { + "name": "Niklas Keller", + "email": "me@kelunik.com" + } + ], + "description": "Rock-solid event loop for concurrent PHP applications.", + "keywords": [ + "async", + "asynchronous", + "concurrency", + "event", + "event-loop", + "non-blocking", + "scheduler" + ], + "support": { + "issues": "https://github.com/revoltphp/event-loop/issues", + "source": "https://github.com/revoltphp/event-loop/tree/v1.0.9" + }, + "time": "2026-05-16T17:55:38+00:00" + }, { "name": "sanmai/di-container", "version": "0.1.17", @@ -7723,6 +7795,84 @@ ], "time": "2026-05-29T05:06:50+00:00" }, + { + "name": "symfony/tui", + "version": "v8.1.6", + "source": { + "type": "git", + "url": "https://github.com/symfony/tui.git", + "reference": "2e141110b41441c6bc97604e6730d90d5066895d" + }, + "dist": { + "type": "zip", + "url": "https://api.github.com/repos/symfony/tui/zipball/2e141110b41441c6bc97604e6730d90d5066895d", + "reference": "2e141110b41441c6bc97604e6730d90d5066895d", + "shasum": "" + }, + "require": { + "php": ">=8.4.1", + "revolt/event-loop": "^1.0", + "symfony/event-dispatcher": "^8.0", + "symfony/string": "^8.0" + }, + "require-dev": { + "league/commonmark": "^2.9", + "tempest/highlight": "^2.16" + }, + "type": "library", + "autoload": { + "psr-4": { + "Symfony\\Component\\Tui\\": "" + }, + "exclude-from-classmap": [ + "/Tests/" + ] + }, + "notification-url": "https://packagist.org/downloads/", + "license": [ + "MIT" + ], + "authors": [ + { + "name": "Fabien Potencier", + "email": "fabien@symfony.com" + }, + { + "name": "Symfony Community", + "homepage": "https://symfony.com/contributors" + } + ], + "description": "Provides a terminal UI framework for building rich, interactive CLI applications in PHP", + "homepage": "https://symfony.com", + "keywords": [ + "cli", + "console", + "terminal", + "tui" + ], + "support": { + "source": "https://github.com/symfony/tui/tree/v8.1.6" + }, + "funding": [ + { + "url": "https://symfony.com/sponsor", + "type": "custom" + }, + { + "url": "https://github.com/fabpot", + "type": "github" + }, + { + "url": "https://github.com/nicolas-grekas", + "type": "github" + }, + { + "url": "https://tidelift.com/funding/github/packagist/symfony/symfony", + "type": "tidelift" + } + ], + "time": "2026-08-27T10:28:13+00:00" + }, { "name": "symfony/var-dumper", "version": "v8.1.0", diff --git a/docs/cli.md b/docs/cli.md index 14741234f..f73ab5e51 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -9,6 +9,7 @@ You can: * Create and delete `databases` * Create, update and delete `schemas` * Manage `subscriptions` +* Monitor and manage `subscriptions` in an interactive dashboard * Inspect your `aggregates`, `events` and `subscribers` * Migrate events from one `store` to another @@ -179,6 +180,7 @@ The boot and run commands are workers and support the following options: | `--sleep` | How many milliseconds the worker waits between two runs. | | `--restart-signal-file` | Stop the worker when this file is touched after it has started. | | `--heartbeat-file` | Touch this file on start and after every run, and remove it when the worker stops. | +| `--stop-when-finished` | Stop the run worker as soon as all events are processed (run command only). | ```bash bin/console event-sourcing:subscription:run --memory-limit=250MB --time-limit=3600 @@ -330,6 +332,90 @@ bin/console event-sourcing:subscription:reactivate --id=welcome_email You can find out more about the [subscription lifecycle](subscription.md#subscription-status). ::: +## Subscription dashboard + +The subscription dashboard is an interactive, full screen terminal UI inspired by k9s. +It shows all subscriptions in a live updating table with their status, position, lag and messages per second, +and lets you run the subscription commands directly on the selected subscriptions. + +* SubscriptionDashboardCommand: `event-sourcing:subscription:dashboard` + +:::experimental +The dashboard is experimental and may change in a minor release. +::: + +The dashboard is built on the [Symfony TUI component](https://symfony.com/doc/current/tui.html), +which is an optional dependency and requires PHP 8.4 or newer. + +```bash +composer require symfony/tui +``` +Register the command with the subscription engine and the store. +The store is optional, it is used to show how far each subscription is behind. + +```php +use Patchlevel\EventSourcing\Console\Command\SubscriptionDashboardCommand; +use Patchlevel\EventSourcing\Store\Store; +use Patchlevel\EventSourcing\Subscription\Engine\SubscriptionEngine; +use Symfony\Component\Console\Application; + +/** + * @var Application $cli + * @var SubscriptionEngine $subscriptionEngine + * @var Store $store + */ +$cli->add( + new SubscriptionDashboardCommand( + $subscriptionEngine, + $store, + ), +); +``` +Like the other subscription commands, you can limit the dashboard to some subscriptions with `--id` and `--group`. +With `--refresh` you set how often the data is reloaded in seconds, +and `--message-limit` sets how many messages a run, boot or rebuild processes at once. + +```bash +bin/console event-sourcing:subscription:dashboard --group=projector --refresh=1 +``` +The actions are executed with the [subscription commands](#subscription-commands) in a separate process, +so the dashboard keeps updating while a long boot or rebuild is running. +This requires the subscription commands to be registered in the same console application. +Otherwise, or with the `--in-process` option, the actions run in the dashboard process and block it until they are done. + +:::note +A run started from the dashboard stops as soon as all events are processed, it does not replace the [run worker](#run). +::: + +### Keyboard shortcuts + +Actions apply to the marked subscriptions, or to the selected one if nothing is marked. +The dashboard only offers actions that fit the status of the subscription, +for example `pause` only for active, booting or failing subscriptions. + +| Key | Action | +|---|---| +| `s` | Setup new subscriptions | +| `b` | Boot subscriptions in booting state | +| `r` | Run active subscriptions | +| `p` | Pause subscriptions | +| `a` | Reactivate subscriptions | +| `f` | Refresh group, run mode and cleanup tasks | +| `t` | Teardown detached subscriptions | +| `ctrl-d` | Remove subscriptions | +| `shift-r` | Rebuild (remove and boot) subscriptions | +| `enter` | Show the details and the error of a subscription | +| `/` | Filter by id, group, status or run mode | +| `space` | Mark or unmark a subscription | +| `esc` | Go back, clear the marks or clear the filter | +| `?` | Show all shortcuts | +| `q` | Quit | + +:::danger +Remove, teardown and rebuild delete the data of the subscribers. +The dashboard asks for a confirmation before it executes them. +::: + ## Inspector commands The inspector commands display the events in your store. diff --git a/docs/subscription.md b/docs/subscription.md index d2388bf9e..16e6ff863 100644 --- a/docs/subscription.md +++ b/docs/subscription.md @@ -1647,5 +1647,6 @@ foreach ($subscriptions as $subscription) { ## Learn more * [How to use CLI commands](cli.md) +* [How to monitor subscriptions with the dashboard](cli.md#subscription-dashboard) * [How to create Messages](message.md) * [How to Test](testing.md) diff --git a/src/Console/Command/SubscriptionDashboardCommand.php b/src/Console/Command/SubscriptionDashboardCommand.php new file mode 100644 index 000000000..ec5ddab03 --- /dev/null +++ b/src/Console/Command/SubscriptionDashboardCommand.php @@ -0,0 +1,139 @@ +addOption( + 'refresh', + null, + InputOption::VALUE_REQUIRED, + 'How often the subscriptions are reloaded in seconds', + 2, + ) + ->addOption( + 'in-process', + null, + InputOption::VALUE_NONE, + 'Execute the actions in the dashboard process instead of starting the subscription commands', + ) + ->addOption( + 'message-limit', + null, + InputOption::VALUE_REQUIRED, + 'How many messages run, boot and rebuild process at once', + 1000, + ); + } + + protected function execute(InputInterface $input, OutputInterface $output): int + { + $io = new OutputStyle($input, $output); + + if (!class_exists(Tui::class)) { + $io->error('The dashboard requires the symfony/tui component (PHP 8.4+). Try running "composer require symfony/tui".'); + + return 1; + } + + if (!$input->isInteractive() && $this->terminal === null) { + $io->error('The dashboard requires an interactive terminal.'); + + return 2; + } + + $refresh = $input->getOption('refresh'); + + $dashboard = new SubscriptionDashboard( + $this->engine, + $this->store, + $this->subscriptionEngineCriteria($input), + is_numeric($refresh) ? max(0.1, (float)$refresh) : 2.0, + $this->actionRunner($input), + $this->terminal, + $this->clock, + ); + + $dashboard->run(); + + return 0; + } + + /** + * Starts the actions as separate processes with the subscription commands of this application, + * so the dashboard keeps rendering. Falls back to the engine if they are not available. + */ + private function actionRunner(InputInterface $input): ActionRunner + { + $messageLimit = InputHelper::nullablePositiveInt($input->getOption('message-limit')); + $application = $this->getApplication(); + $argv = $_SERVER['argv'] ?? null; + $script = is_array($argv) ? $argv[0] ?? null : null; + + if ( + InputHelper::bool($input->getOption('in-process')) + || $application === null + || !is_string($script) + || !is_file($script) + || array_filter(ProcessActionRunner::COMMANDS, static fn (string $name) => !$application->has($name)) !== [] + ) { + return new InProcessActionRunner($this->engine, $messageLimit); + } + + $console = [PHP_BINARY, $script]; + $env = $input->getParameterOption(['--env', '-e'], null, true); + + if (is_string($env)) { + $console[] = '--env=' . $env; + } + + return new ProcessActionRunner($console, $messageLimit); + } +} diff --git a/src/Console/Command/SubscriptionRunCommand.php b/src/Console/Command/SubscriptionRunCommand.php index 862f3e9c7..e3548507c 100644 --- a/src/Console/Command/SubscriptionRunCommand.php +++ b/src/Console/Command/SubscriptionRunCommand.php @@ -4,6 +4,7 @@ namespace Patchlevel\EventSourcing\Console\Command; +use Closure; use Patchlevel\EventSourcing\Console\InputHelper; use Patchlevel\EventSourcing\Store\ListenableStore; use Patchlevel\EventSourcing\Store\Store; @@ -88,6 +89,12 @@ protected function configure(): void null, InputOption::VALUE_NONE, 'rebuild (remove & boot) subscriptions before run', + ) + ->addOption( + 'stop-when-finished', + null, + InputOption::VALUE_NONE, + 'Stop the worker as soon as all messages are processed', ); } @@ -99,6 +106,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int $timeLimit = InputHelper::nullablePositiveInt($input->getOption('time-limit')); $sleep = InputHelper::positiveIntOrZero($input->getOption('sleep')); $rebuild = InputHelper::bool($input->getOption('rebuild')); + $stopWhenFinished = InputHelper::bool($input->getOption('stop-when-finished')); $restartSignalFile = InputHelper::nullableString($input->getOption('restart-signal-file')); $heartbeatFile = InputHelper::nullableString($input->getOption('heartbeat-file')); @@ -108,13 +116,19 @@ protected function execute(InputInterface $input, OutputInterface $output): int $logger = new ConsoleLogger($output); $worker = DefaultWorker::create( - function () use ($criteria, $messageLimit, $sleep): bool { + function (Closure $stop) use ($criteria, $messageLimit, $sleep, $stopWhenFinished): bool { $result = $this->engine->execute(new Run($criteria->ids, $criteria->groups, $messageLimit)); if ($result instanceof ProcessedResult && !$result->finished) { return true; } + if ($stopWhenFinished) { + $stop(); + + return false; + } + if ($this->store instanceof ListenableStore) { $this->store->wait($sleep); } diff --git a/src/Console/Tui/ActionRunner/ActionRunner.php b/src/Console/Tui/ActionRunner/ActionRunner.php new file mode 100644 index 000000000..b2d899bdd --- /dev/null +++ b/src/Console/Tui/ActionRunner/ActionRunner.php @@ -0,0 +1,21 @@ + $ids + * @param Closure(list): void $onFinish + */ + public function run(SubscriptionAction $action, array $ids, Closure $onFinish): void; +} diff --git a/src/Console/Tui/ActionRunner/InProcessActionRunner.php b/src/Console/Tui/ActionRunner/InProcessActionRunner.php new file mode 100644 index 000000000..a84281d2c --- /dev/null +++ b/src/Console/Tui/ActionRunner/InProcessActionRunner.php @@ -0,0 +1,49 @@ + $ids + * @param Closure(list): void $onFinish + */ + public function run(SubscriptionAction $action, array $ids, Closure $onFinish): void + { + $errors = []; + + try { + foreach ($action->commands($ids, $this->messageLimit) as $command) { + foreach ($this->engine->execute($command)->errors as $error) { + $errors[] = sprintf('%s: %s', $error->subscriptionId, $error->message); + } + } + } catch (Throwable $e) { + $errors[] = $e->getMessage(); + } + + $onFinish($errors); + } +} diff --git a/src/Console/Tui/ActionRunner/ProcessActionRunner.php b/src/Console/Tui/ActionRunner/ProcessActionRunner.php new file mode 100644 index 000000000..e99634230 --- /dev/null +++ b/src/Console/Tui/ActionRunner/ProcessActionRunner.php @@ -0,0 +1,173 @@ + $console the command to start the console, e.g. [PHP_BINARY, 'bin/console'] + * @param positive-int|null $messageLimit + */ + public function __construct( + private readonly array $console, + private readonly int|null $messageLimit = null, + private readonly float $pollInterval = 0.1, + ) { + } + + /** + * @param list $ids + * @param Closure(list): void $onFinish + */ + public function run(SubscriptionAction $action, array $ids, Closure $onFinish): void + { + $options = [ + ...array_map(static fn (string $id) => '--id=' . $id, $ids), + '--no-interaction', + ]; + + $queue = array_map( + static fn (array $arguments) => [...$arguments, ...$options], + $this->arguments($action), + ); + + $this->next($queue, $onFinish); + } + + /** @return list> */ + private function arguments(SubscriptionAction $action): array + { + $limit = $this->messageLimit === null ? [] : ['--message-limit=' . $this->messageLimit]; + + return match ($action) { + SubscriptionAction::Setup => [['event-sourcing:subscription:setup']], + SubscriptionAction::Boot => [['event-sourcing:subscription:boot', ...$limit]], + SubscriptionAction::Run => [['event-sourcing:subscription:run', '--stop-when-finished', '--sleep=0', ...$limit]], + SubscriptionAction::Pause => [['event-sourcing:subscription:pause']], + SubscriptionAction::Reactivate => [['event-sourcing:subscription:reactivate']], + SubscriptionAction::Refresh => [['event-sourcing:subscription:refresh']], + SubscriptionAction::Teardown => [['event-sourcing:subscription:teardown']], + SubscriptionAction::Remove => [['event-sourcing:subscription:remove', '--force']], + SubscriptionAction::Rebuild => [ + ['event-sourcing:subscription:remove', '--force'], + ['event-sourcing:subscription:boot', '--setup', ...$limit], + ], + }; + } + + /** + * @param list> $queue + * @param Closure(list): void $onFinish + */ + private function next(array $queue, Closure $onFinish): void + { + $arguments = array_shift($queue); + + if ($arguments === null) { + $onFinish([]); + + return; + } + + $output = tmpfile(); + + if ($output === false) { + $onFinish(['Could not create a temporary file for the command output']); + + return; + } + + $process = proc_open([...$this->console, ...$arguments], [0 => ['pipe', 'r'], 1 => $output, 2 => $output], $pipes); + + if (!is_resource($process)) { + fclose($output); + $onFinish([sprintf('Could not start "%s"', $arguments[0])]); + + return; + } + + fclose($pipes[0]); + + EventLoop::repeat($this->pollInterval, function (string $callbackId) use ($process, $output, $arguments, $queue, $onFinish): void { + $status = proc_get_status($process); + + if ($status['running']) { + return; + } + + EventLoop::cancel($callbackId); + proc_close($process); + + rewind($output); + $content = (string)stream_get_contents($output); + fclose($output); + + if ($status['exitcode'] !== 0) { + $onFinish([ + sprintf( + '"%s" exited with code %d%s', + $arguments[0], + $status['exitcode'], + self::lastLine($content) === '' ? '' : ': ' . self::lastLine($content), + ), + ]); + + return; + } + + $this->next($queue, $onFinish); + }); + } + + private static function lastLine(string $output): string + { + $lines = array_values(array_filter( + array_map(trim(...), explode("\n", $output)), + static fn (string $line) => $line !== '', + )); + + return $lines === [] ? '' : (string)end($lines); + } +} diff --git a/src/Console/Tui/SubscriptionAction.php b/src/Console/Tui/SubscriptionAction.php new file mode 100644 index 000000000..e760d059b --- /dev/null +++ b/src/Console/Tui/SubscriptionAction.php @@ -0,0 +1,134 @@ + 's', + self::Boot => 'b', + self::Run => 'r', + self::Pause => 'p', + self::Reactivate => 'a', + self::Refresh => 'f', + self::Teardown => 't', + self::Remove => 'ctrl+d', + self::Rebuild => 'shift+r', + }; + } + + /** The key as shown to the user. */ + public function keyLabel(): string + { + return match ($this) { + self::Remove => 'ctrl-d', + self::Rebuild => 'shift-r', + default => $this->key(), + }; + } + + public function label(): string + { + return match ($this) { + self::Setup => 'Setup', + self::Boot => 'Boot', + self::Run => 'Run', + self::Pause => 'Pause', + self::Reactivate => 'Reactivate', + self::Refresh => 'Refresh', + self::Teardown => 'Teardown', + self::Remove => 'Remove', + self::Rebuild => 'Rebuild', + }; + } + + public function description(): string + { + return match ($this) { + self::Setup => 'Set up new subscriptions', + self::Boot => 'Boot subscriptions that are in booting state', + self::Run => 'Process new messages for active subscriptions', + self::Pause => 'Pause active, booting or failing subscriptions', + self::Reactivate => 'Reactivate paused, finished, detached or failing subscriptions', + self::Refresh => 'Sync group, run mode and cleanup tasks from the subscriber', + self::Teardown => 'Tear down detached subscriptions', + self::Remove => 'Remove subscriptions and call their teardown', + self::Rebuild => 'Remove and boot subscriptions again', + }; + } + + public function needsConfirmation(): bool + { + return in_array($this, [self::Teardown, self::Remove, self::Rebuild], true); + } + + /** + * Mirrors the statuses the engine handlers act on, + * so that only actions that will actually do something are offered. + */ + public function supports(Subscription $subscription): bool + { + $statuses = match ($this) { + self::Setup => [Status::New], + self::Boot => [Status::Booting], + self::Run => [Status::Active], + self::Pause => [Status::Active, Status::Booting, Status::Error], + self::Reactivate => [Status::Error, Status::Failed, Status::Detached, Status::Paused, Status::Finished], + self::Teardown => [Status::Detached], + self::Refresh, self::Remove, self::Rebuild => null, + }; + + return $statuses === null || in_array($subscription->status(), $statuses, true); + } + + /** + * @param list $ids + * @param positive-int|null $messageLimit + * + * @return list + */ + public function commands(array $ids, int|null $messageLimit = null): array + { + return match ($this) { + self::Setup => [new Setup($ids)], + self::Boot => [new Boot($ids, null, $messageLimit)], + self::Run => [new Run($ids, null, $messageLimit)], + self::Pause => [new Pause($ids)], + self::Reactivate => [new Reactivate($ids)], + self::Refresh => [new Refresh($ids)], + self::Teardown => [new Teardown($ids)], + self::Remove => [new Remove($ids)], + self::Rebuild => [new Remove($ids), new Boot($ids, null, $messageLimit)], + }; + } +} diff --git a/src/Console/Tui/SubscriptionDashboard.php b/src/Console/Tui/SubscriptionDashboard.php new file mode 100644 index 000000000..1272e1311 --- /dev/null +++ b/src/Console/Tui/SubscriptionDashboard.php @@ -0,0 +1,741 @@ + */ + private array $pendingTargets = []; + + private int|null $head = null; + private readonly Throughput $headThroughput; + private DateTimeImmutable|null $lastReload = null; + private int $flashUntil = 0; + + private string|null $running = null; + private string|null $spinnerTimer = null; + private int $spinnerFrame = 0; + + public function __construct( + private readonly SubscriptionEngine $engine, + private readonly Store|null $store = null, + private readonly SubscriptionEngineCriteria|null $criteria = null, + private readonly float $refreshInterval = 2.0, + ActionRunner|null $runner = null, + TerminalInterface|null $terminal = null, + private readonly ClockInterface|null $clock = null, + ) { + $this->runner = $runner ?? new InProcessActionRunner($engine); + $this->keybindings = self::keybindings(); + $this->tui = new Tui(self::styleSheet(), $terminal ?? new Terminal(), $this->keybindings); + + $this->headThroughput = new Throughput(); + $this->table = new SubscriptionTableView(); + $this->describe = new TextView('Describe'); + $this->help = new TextView('Help', $this->helpLines()); + + $this->header = new HeaderWidget(); + $this->panel = new PanelWidget($this->table); + $this->footer = new FooterWidget(); + + $this->tui->add($this->header); + $this->tui->add($this->panel); + $this->tui->add($this->footer); + + $this->tui->addListener($this->onInput(...)); + + $this->reload(); + } + + public function run(): void + { + $terminal = $this->tui->getTerminal(); + $terminal->write("\x1b[?1049h\x1b[2J\x1b[H"); + + $timer = EventLoop::repeat($this->refreshInterval, function (): void { + $this->reload(); + }); + + try { + $this->tui->requestRender(true); + $this->tui->run(); + } finally { + EventLoop::cancel($timer); + $this->stopSpinner(); + $terminal->write("\x1b[?1049l"); + } + } + + public function reload(): void + { + try { + $subscriptions = $this->engine->subscriptions($this->criteria); + $this->head = $this->head(); + } catch (Throwable $e) { + $this->flash(sprintf('Failed to load subscriptions: %s', $e->getMessage()), Theme::DANGER); + $this->tui->requestRender(); + + return; + } + + $this->lastReload = $this->now(); + $this->headThroughput->record('head', $this->head, $this->lastReload); + $this->table->update($subscriptions, $this->head, $this->lastReload); + + if ($this->flashUntil > 0 && $this->flashUntil < $this->lastReload->getTimestamp()) { + $this->flashUntil = 0; + $this->footer->setMessage(''); + } + + if ($this->describedId !== null) { + $this->updateDescribe(); + } + + $this->updateChrome(); + } + + private function onInput(InputEvent $event): void + { + $event->stopPropagation(); + $data = $event->getData(); + + if ($this->matches($data, 'force_quit')) { + $this->tui->stop(); + + return; + } + + if ($this->panel->hasDialog()) { + $this->handleDialogInput($data); + } elseif ($this->filterEditing) { + $this->handleFilterInput($data); + } else { + $this->handleInput($data); + } + + $this->updateChrome(); + } + + private function handleDialogInput(string $data): void + { + if ($this->matches($data, 'confirm')) { + $action = $this->pendingAction; + $targets = $this->pendingTargets; + $this->closeDialog(); + + if ($action !== null) { + $this->execute($action, $targets); + } + + return; + } + + if (!$this->matches($data, 'cancel')) { + return; + } + + $this->closeDialog(); + $this->flash('Cancelled'); + } + + private function handleFilterInput(string $data): void + { + if ($this->matches($data, 'filter_apply')) { + $this->filterEditing = false; + + return; + } + + if ($this->matches($data, 'back')) { + $this->filterEditing = false; + $this->table->changeFilter(''); + + return; + } + + if ($this->matches($data, 'filter_backspace')) { + $filter = $this->table->filter(); + $this->table->changeFilter(mb_substr($filter, 0, -1)); + + return; + } + + if ($this->hasControlCharacters($data)) { + return; + } + + $this->table->changeFilter($this->table->filter() . $data); + } + + private function handleInput(string $data): void + { + $view = $this->panel->view(); + + if ($this->matches($data, 'quit')) { + $this->tui->stop(); + + return; + } + + if ($this->matches($data, 'back')) { + $this->back(); + + return; + } + + if ($this->matches($data, 'up')) { + $view->moveBy(-1); + } elseif ($this->matches($data, 'down')) { + $view->moveBy(1); + } elseif ($this->matches($data, 'page_up')) { + $view->pageUp(); + } elseif ($this->matches($data, 'page_down')) { + $view->pageDown(); + } elseif ($this->matches($data, 'start')) { + $view->moveToStart(); + } elseif ($this->matches($data, 'end')) { + $view->moveToEnd(); + } elseif ($this->matches($data, 'help')) { + $this->describedId = null; + $this->panel->show($this->help); + } elseif ($this->matches($data, 'reload')) { + $this->reload(); + $this->flash('Reloaded'); + } elseif ($view === $this->table) { + $this->handleTableInput($data); + } + + foreach (SubscriptionAction::cases() as $action) { + if (!$this->matches($data, 'action_' . $action->value)) { + continue; + } + + $this->trigger($action); + + return; + } + + $this->panel->invalidate(); + } + + private function handleTableInput(string $data): void + { + if ($this->matches($data, 'filter')) { + $this->filterEditing = true; + } elseif ($this->matches($data, 'mark')) { + $this->table->toggleMark(); + } elseif ($this->matches($data, 'describe')) { + $selected = $this->table->selected(); + + if ($selected === null) { + return; + } + + $this->describedId = $selected->id(); + $this->updateDescribe(); + $this->describe->moveToStart(); + $this->panel->show($this->describe); + } + } + + private function back(): void + { + if ($this->panel->view() !== $this->table) { + $this->describedId = null; + $this->panel->show($this->table); + + return; + } + + if ($this->table->hasMarks()) { + $this->table->clearMarks(); + + return; + } + + $this->table->changeFilter(''); + } + + private function trigger(SubscriptionAction $action): void + { + if ($this->running !== null) { + $this->flash(sprintf('Wait until "%s" is done', $this->running), Theme::WARNING); + + return; + } + + $targets = $this->targets(); + + if ($targets === []) { + $this->flash('No subscription selected', Theme::WARNING); + + return; + } + + $supported = array_values(array_filter($targets, $action->supports(...))); + + if ($supported === []) { + $this->flash( + count($targets) === 1 + ? sprintf('%s is not possible for subscription "%s" with status "%s"', $action->label(), $targets[0]->id(), $targets[0]->status()->value) + : sprintf('%s is not possible for any of the marked subscriptions', $action->label()), + Theme::WARNING, + ); + + return; + } + + if (!$action->needsConfirmation()) { + $this->execute($action, $supported); + + return; + } + + $this->pendingAction = $action; + $this->pendingTargets = $supported; + + $this->panel->openDialog( + $action->label(), + [ + Theme::color(Theme::TEXT, sprintf('%s %s?', $action->label(), $this->describeTargets($supported))), + Theme::color(Theme::MUTED, $action->description()), + '', + Theme::color('1;38;5;255;48;5;167', sprintf(' y %s ', $action->label())) + . ' ' . Theme::color(Theme::TEXT . ';' . Theme::PILL_BACKGROUND, ' n Cancel '), + ], + Theme::DANGER, + ); + } + + /** @param list $targets */ + private function execute(SubscriptionAction $action, array $targets): void + { + $ids = array_map(static fn (Subscription $subscription) => $subscription->id(), $targets); + $label = sprintf('%s %s', $action->label(), $this->describeTargets($targets)); + + $this->running = $label; + $this->showRunning(); + + // render before the runner starts, an in process runner blocks until it is done + $this->tui->requestRender(); + $this->tui->processRender(); + + $this->runner->run($action, $ids, function (array $errors) use ($action, $label): void { + $this->running = null; + $this->stopSpinner(); + $this->table->clearMarks(); + $this->reload(); + + if ($errors === []) { + $this->flash(sprintf('%s done', $label), Theme::SUCCESS); + } else { + $this->flash( + sprintf('%s failed: %s', $action->label(), $errors[0]) . (count($errors) > 1 ? sprintf(' (+%d more)', count($errors) - 1) : ''), + Theme::DANGER, + ); + } + + $this->updateChrome(); + }); + + if ($this->running === null) { + return; + } + + $this->spinnerTimer = EventLoop::repeat(0.08, function (): void { + $this->spinnerFrame++; + $this->showRunning(); + $this->tui->requestRender(); + }); + } + + private function showRunning(): void + { + if ($this->running === null) { + return; + } + + $this->flashUntil = 0; + $this->footer->setMessage( + sprintf('%s %s...', self::SPINNER[$this->spinnerFrame % count(self::SPINNER)], $this->running), + Theme::ACCENT, + ); + } + + private function stopSpinner(): void + { + if ($this->spinnerTimer === null) { + return; + } + + EventLoop::cancel($this->spinnerTimer); + $this->spinnerTimer = null; + } + + /** @return list */ + private function targets(): array + { + if ($this->panel->view() === $this->table) { + return $this->table->targets(); + } + + $described = $this->described(); + + return $described === null ? [] : [$described]; + } + + private function described(): Subscription|null + { + foreach ($this->table->subscriptions() as $subscription) { + if ($subscription->id() === $this->describedId) { + return $subscription; + } + } + + return null; + } + + private function updateDescribe(): void + { + $subscription = $this->described(); + + if ($subscription === null) { + $this->flash(sprintf('Subscription "%s" no longer exists', (string)$this->describedId), Theme::WARNING); + $this->describedId = null; + $this->panel->show($this->table); + + return; + } + + $this->describe->update( + sprintf('Describe(%s)', $subscription->id()), + SubscriptionFormatter::describe( + $subscription, + $this->head, + $this->lastReload ?? $this->now(), + $this->table->rate($subscription->id()), + ), + ); + } + + private function closeDialog(): void + { + $this->pendingAction = null; + $this->pendingTargets = []; + $this->panel->closeDialog(); + } + + private function updateChrome(): void + { + $view = $this->panel->view(); + + $crumbs = ['subscriptions']; + + if ($view === $this->describe && $this->describedId !== null) { + $crumbs[] = $this->describedId; + } elseif ($view === $this->help) { + $crumbs[] = 'help'; + } + + $meta = []; + + if ($this->scope() !== null) { + $meta[] = Theme::color(Theme::HIGHLIGHT, (string)$this->scope()); + } + + $headRate = $this->headThroughput->rate('head'); + $meta[] = Theme::color(Theme::MUTED, 'head ') . Theme::color(Theme::TEXT, $this->head === null ? '–' : Theme::number($this->head)) + . ($headRate === null || $headRate < 0.05 ? '' : Theme::color(Theme::MUTED, sprintf(' +%s/s', Theme::rate($headRate)))); + $meta[] = Theme::color(Theme::MUTED, sprintf('⟳ %ss', $this->refreshInterval)); + $meta[] = Theme::color(Theme::MUTED, $this->lastReload?->format('H:i:s') ?? '–'); + + $this->header->setCrumbs($crumbs); + $this->header->setMeta($meta); + $this->header->setSummary($this->summary($this->table->subscriptions())); + + $this->footer->setHints($this->hints()); + $this->footer->setPrompt($this->filterEditing ? $this->table->filter() : null); + $this->panel->invalidate(); + $this->tui->requestRender(); + } + + /** @return list */ + private function hints(): array + { + if ($this->panel->hasDialog()) { + return [['y', 'Confirm'], ['n', 'Cancel']]; + } + + if ($this->filterEditing) { + return [['enter', 'Apply filter'], ['esc', 'Clear filter']]; + } + + $hints = []; + $targets = $this->targets(); + + foreach (SubscriptionAction::cases() as $action) { + foreach ($targets as $target) { + if ($action->supports($target)) { + $hints[] = [$action->keyLabel(), strtolower($action->label())]; + + break; + } + } + } + + if ($this->panel->view() === $this->table) { + $hints[] = ['enter', 'describe']; + $hints[] = ['/', 'filter']; + $hints[] = ['space', 'mark']; + } else { + $hints[] = ['esc', 'back']; + } + + $hints[] = ['?', 'help']; + $hints[] = ['q', 'quit']; + + return $hints; + } + + /** @param list $subscriptions */ + private function summary(array $subscriptions): string + { + $counts = []; + + foreach ($subscriptions as $subscription) { + $counts[$subscription->status()->value] = ($counts[$subscription->status()->value] ?? 0) + 1; + } + + $parts = [Theme::bold(Theme::TEXT, (string)count($subscriptions)) . Theme::color(Theme::MUTED, ' subscriptions')]; + + foreach (Status::cases() as $status) { + if (!isset($counts[$status->value])) { + continue; + } + + $parts[] = Theme::color( + Theme::statusColor($status), + sprintf('%s %d %s', Theme::statusIcon($status), $counts[$status->value], $status->value), + ); + } + + return implode(' ', $parts); + } + + private function scope(): string|null + { + $parts = []; + + if ($this->criteria?->ids !== null) { + $parts[] = 'id=' . implode(',', $this->criteria->ids); + } + + if ($this->criteria?->groups !== null) { + $parts[] = 'group=' . implode(',', $this->criteria->groups); + } + + return $parts === [] ? null : implode(' ', $parts); + } + + /** @param list $targets */ + private function describeTargets(array $targets): string + { + if (count($targets) === 1) { + return sprintf('"%s"', $targets[0]->id()); + } + + return sprintf('%d subscriptions', count($targets)); + } + + private function flash(string $message, string $color = Theme::MUTED): void + { + $this->flashUntil = $this->now()->getTimestamp() + self::FLASH_SECONDS; + $this->footer->setMessage($message, $color); + } + + private function head(): int|null + { + if ($this->store === null) { + return null; + } + + $stream = $this->store->load(null, 1, null, true); + + try { + $message = $stream->current(); + + if ($message === null) { + return 0; + } + + if ($message->hasHeader(IndexHeader::class)) { + return $message->header(IndexHeader::class)->index; + } + + return $stream->index() ?? 0; + } finally { + $stream->close(); + } + } + + private function now(): DateTimeImmutable + { + return $this->clock?->now() ?? new DateTimeImmutable(); + } + + private function matches(string $data, string $action): bool + { + return $this->keybindings->matches($data, $action); + } + + private function hasControlCharacters(string $data): bool + { + for ($i = 0, $length = strlen($data); $i < $length; $i++) { + $code = ord($data[$i]); + + if ($code < 32 || $code === 0x7F) { + return true; + } + } + + return false; + } + + /** @return list */ + private function helpLines(): array + { + $line = static fn (string $key, string $description) => ' ' . Theme::keyHint($key, '') + . str_repeat(' ', max(1, 12 - AnsiUtils::visibleWidth($key))) . Theme::color(Theme::TEXT, $description); + + $lines = [Theme::bold(Theme::ACCENT, 'Actions'), '']; + + foreach (SubscriptionAction::cases() as $action) { + $lines[] = $line($action->keyLabel(), $action->description()); + } + + return [ + ...$lines, + '', + Theme::color(Theme::MUTED, ' Actions apply to the marked subscriptions, or to the selected one if nothing is marked.'), + '', + Theme::bold(Theme::ACCENT, 'Navigation'), + '', + $line('↑ ↓ j k', 'Move the selection'), + $line('pgup pgdn', 'Move one page'), + $line('g G', 'Jump to the first or last entry'), + $line('enter d', 'Describe the selected subscription'), + $line('space', 'Mark or unmark the selected subscription'), + $line('/', 'Filter by id, group, status or run mode'), + $line('esc', 'Go back, clear marks or clear the filter'), + $line('ctrl-r', 'Reload now'), + $line('?', 'Show this help'), + $line('q', 'Quit'), + ]; + } + + private static function keybindings(): Keybindings + { + $bindings = [ + 'quit' => ['q'], + 'force_quit' => ['ctrl+c'], + 'back' => [Key::ESCAPE], + 'up' => [Key::UP, 'k'], + 'down' => [Key::DOWN, 'j'], + 'page_up' => [Key::PAGE_UP], + 'page_down' => [Key::PAGE_DOWN], + 'start' => [Key::HOME, 'g'], + 'end' => [Key::END, 'shift+g'], + 'describe' => [Key::ENTER, 'd'], + 'filter' => ['/'], + 'filter_apply' => [Key::ENTER], + 'filter_backspace' => [Key::BACKSPACE], + 'mark' => [Key::SPACE], + 'reload' => ['ctrl+r'], + 'help' => ['?'], + 'confirm' => ['y', Key::ENTER], + 'cancel' => ['n', Key::ESCAPE], + ]; + + foreach (SubscriptionAction::cases() as $action) { + $bindings['action_' . $action->value] = [$action->key()]; + } + + return new Keybindings($bindings); + } + + private static function styleSheet(): StyleSheet + { + return new StyleSheet([ + ':root' => new Style(direction: Direction::Vertical, gap: 0, verticalAlign: VerticalAlign::Top), + ]); + } +} diff --git a/src/Console/Tui/SubscriptionFormatter.php b/src/Console/Tui/SubscriptionFormatter.php new file mode 100644 index 000000000..37f69c353 --- /dev/null +++ b/src/Console/Tui/SubscriptionFormatter.php @@ -0,0 +1,241 @@ +position(); + + if ($head === null || $position === null) { + return null; + } + + return max(0, $head - $position); + } + + public static function age(DateTimeImmutable|null $time, DateTimeImmutable $now): string + { + if ($time === null) { + return '-'; + } + + return self::duration(max(0, $now->getTimestamp() - $time->getTimestamp())); + } + + public static function duration(int $seconds): string + { + return match (true) { + $seconds < 60 => $seconds . 's', + $seconds < 3600 => intdiv($seconds, 60) . 'm', + $seconds < 86400 => intdiv($seconds, 3600) . 'h', + default => intdiv($seconds, 86400) . 'd', + }; + } + + public static function runMode(Subscription $subscription): string + { + return str_replace('_', ' ', $subscription->runMode()->value); + } + + /** @return list */ + public static function describe(Subscription $subscription, int|null $head, DateTimeImmutable $now, float|null $rate = null): array + { + $error = $subscription->subscriptionError(); + $status = Theme::status($subscription->status()); + + if ($error !== null) { + $status .= Theme::color(Theme::MUTED, sprintf(' was %s', $error->previousStatus->value)); + } + + $lastSavedAt = $subscription->lastSavedAt(); + $cleanupTasks = $subscription->cleanupTasks(); + + $lines = [ + ...self::section('Overview'), + self::field('ID', Theme::bold(Theme::TEXT, $subscription->id())), + self::field('Group', $subscription->group()), + self::field('Run mode', self::runMode($subscription)), + self::field('Status', $status), + self::field('Progress', self::progress($subscription, $head)), + self::field('Throughput', self::throughput($subscription, $head, $rate)), + self::field('Retries', $subscription->retryAttempt() === 0 + ? Theme::color(Theme::SUBTLE, 'none') + : Theme::color(Theme::WARNING, (string)$subscription->retryAttempt())), + self::field('Last saved', $lastSavedAt === null + ? Theme::color(Theme::SUBTLE, '–') + : $lastSavedAt->format('Y-m-d H:i:s') . Theme::color(Theme::MUTED, sprintf(' %s ago', self::age($lastSavedAt, $now)))), + self::field('Cleanup tasks', $cleanupTasks === null ? Theme::color(Theme::SUBTLE, 'none') : (string)count($cleanupTasks)), + ]; + + if ($error === null) { + return $lines; + } + + $lines[] = ''; + $lines = [...$lines, ...self::section('Error', Theme::DANGER)]; + + if ($error->errorContext === null || $error->errorContext === []) { + $lines[] = Theme::color(Theme::DANGER, '▎ ') . Theme::color(Theme::TEXT, $error->errorMessage); + + return $lines; + } + + foreach ($error->errorContext as $context) { + foreach (self::context($context) as $line) { + $lines[] = $line; + } + } + + return $lines; + } + + private static function progress(Subscription $subscription, int|null $head): string + { + $position = $subscription->position(); + + if ($position === null) { + return Theme::color(Theme::SUBTLE, 'not started'); + } + + if ($head === null) { + return Theme::number($position); + } + + $lag = self::lag($subscription, $head) ?? 0; + + return Theme::progressBar($position, $head, 24) + . ' ' . Theme::color(Theme::TEXT, Theme::percent($position, $head)) + . Theme::color(Theme::MUTED, sprintf(' %s / %s', Theme::number($position), Theme::number($head))) + . ($lag > 0 ? Theme::color(Theme::WARNING, sprintf(' lag %s', Theme::number($lag))) : ''); + } + + private static function throughput(Subscription $subscription, int|null $head, float|null $rate): string + { + if ($rate === null) { + return Theme::color(Theme::SUBTLE, '–'); + } + + $lag = self::lag($subscription, $head) ?? 0; + + return Theme::color(Theme::TEXT, Theme::rate($rate)) . Theme::color(Theme::MUTED, ' msg/s') + . ($lag > 0 && $rate >= 0.05 + ? Theme::color(Theme::MUTED, sprintf(' ~%s left', self::duration((int)ceil($lag / $rate)))) + : ''); + } + + /** @return list */ + private static function section(string $title, string $color = Theme::ACCENT): array + { + return [ + Theme::bold($color, $title), + '', + ]; + } + + /** + * @param Context $context + * + * @return list + */ + private static function context(array $context): array + { + $bar = Theme::color(Theme::DANGER, '▎ '); + + $lines = [ + $bar . Theme::bold(Theme::TEXT, $context['class']), + $bar . Theme::color(Theme::TEXT, $context['message']), + $bar . Theme::color(Theme::MUTED, sprintf('%s:%d', $context['file'], $context['line'])), + '', + Theme::color(Theme::MUTED, 'Stack trace'), + '', + ]; + + foreach ($context['trace'] as $index => $trace) { + $lines[] = Theme::color(Theme::SUBTLE, sprintf('%3d ', $index)) . Theme::color(Theme::TEXT, self::call($trace)); + + if (!isset($trace['file'])) { + continue; + } + + $lines[] = ' ' . Theme::color(Theme::SUBTLE, sprintf('%s:%s', $trace['file'], $trace['line'] ?? '?')); + } + + $lines[] = ''; + + return $lines; + } + + /** @param Trace $trace */ + private static function call(array $trace): string + { + $function = $trace['function'] ?? '{main}'; + + if (isset($trace['class'])) { + $function = self::shortClass($trace['class']) . ($trace['type'] ?? '::') . $function; + } + + $args = []; + + foreach ($trace['args'] ?? [] as $arg) { + $args[] = self::argument($arg); + } + + return $function . '(' . implode(', ', $args) . ')'; + } + + private static function argument(mixed $arg): string + { + if (!is_array($arg) || !isset($arg[0]) || !is_scalar($arg[0])) { + return '?'; + } + + $type = (string)$arg[0]; + $value = isset($arg[1]) && is_scalar($arg[1]) ? (string)$arg[1] : ''; + + return match ($type) { + 'object' => self::shortClass($value), + 'string' => '"' . $value . '"', + 'integer', 'float' => $value, + 'boolean' => $value ? 'true' : 'false', + default => $type, + }; + } + + private static function shortClass(string $class): string + { + $position = strrpos($class, '\\'); + + return $position === false ? $class : substr($class, $position + 1); + } + + private static function field(string $label, string $value): string + { + return Theme::color(Theme::MUTED, str_pad($label, 16)) . Theme::color(Theme::TEXT, $value); + } +} diff --git a/src/Console/Tui/Theme.php b/src/Console/Tui/Theme.php new file mode 100644 index 000000000..a25527f2e --- /dev/null +++ b/src/Console/Tui/Theme.php @@ -0,0 +1,252 @@ + '38;5;80', + Status::Booting => self::WARNING, + Status::Active => self::SUCCESS, + Status::Paused => self::MUTED, + Status::Finished => '38;5;75', + Status::Detached => self::HIGHLIGHT, + Status::Error => '38;5;215', + Status::Failed => self::DANGER, + }; + } + + public static function statusIcon(Status $status): string + { + return match ($status) { + Status::New => '○', + Status::Booting => '◐', + Status::Active => '●', + Status::Paused => '‖', + Status::Finished => '✓', + Status::Detached => '◌', + Status::Error, Status::Failed => '✗', + }; + } + + public static function status(Status $status): string + { + return self::color(self::statusColor($status), self::statusIcon($status) . ' ' . $status->value); + } + + /** A key hint like in modern tuis: the key on a subtle background, followed by the label. */ + public static function keyHint(string $key, string $label): string + { + return self::color('1;' . self::TEXT . ';' . self::PILL_BACKGROUND, ' ' . $key . ' ') + . ' ' . self::color(self::MUTED, $label); + } + + public static function number(int $number): string + { + return number_format($number, 0, '.', ','); + } + + /** Messages per second, with one decimal for small rates. */ + public static function rate(float $rate): string + { + return $rate < 10 ? sprintf('%.1f', $rate) : self::number((int)round($rate)); + } + + public static function progressBar(int $position, int $head, int $width): string + { + $ratio = $head <= 0 ? 1.0 : min(1.0, max(0.0, $position / $head)); + $filled = (int)round($ratio * $width); + + if ($filled === $width && $position < $head) { + $filled = $width - 1; + } + + $color = $ratio >= 1.0 ? self::SUCCESS : self::ACCENT; + + return self::color($color, str_repeat('━', $filled)) + . self::color(self::BORDER, str_repeat('━', $width - $filled)); + } + + public static function percent(int $position, int $head): string + { + if ($head <= 0 || $position >= $head) { + return '100%'; + } + + return sprintf('%d%%', (int)($position / $head * 100)); + } + + /** Truncates (with ellipsis) or pads the text to exactly the given width. */ + public static function fit(string $text, int $width): string + { + if ($width <= 0) { + return ''; + } + + return AnsiUtils::truncateToWidth($text, $width, '…', true); + } + + public static function fitRight(string $text, int $width): string + { + if ($width <= 0) { + return ''; + } + + $visible = AnsiUtils::visibleWidth($text); + + if ($visible >= $width) { + return self::fit($text, $width); + } + + return str_repeat(' ', $width - $visible) . $text; + } + + /** Places the right text at the end of the line, truncating the left text if needed. */ + public static function spread(string $left, string $right, int $width): string + { + $rightWidth = AnsiUtils::visibleWidth($right); + + if ($rightWidth >= $width) { + return self::fit($right, $width); + } + + return self::fit($left, $width - $rightWidth) . $right; + } + + /** + * Draws a box with rounded corners, a title on the left and an optional label on the right. + * + * @param list $lines + * + * @return list + */ + public static function frame( + string $title, + array $lines, + int $columns, + int $rows, + string $label = '', + string $borderColor = self::BORDER, + ): array { + if ($rows <= 0 || $columns <= 0) { + return []; + } + + if ($columns < 8 || $rows < 2) { + return array_fill(0, $rows, str_repeat(' ', $columns)); + } + + $innerWidth = $columns - 4; + $innerRows = $rows - 2; + + $title = ' ' . AnsiUtils::truncateToWidth($title, max(0, $columns - 8), '…') . "\e[0m "; + $label = $label === '' ? '' : ' ' . $label . "\e[0m "; + + if (AnsiUtils::visibleWidth($title) + AnsiUtils::visibleWidth($label) + 5 > $columns) { + $label = ''; + } + + $fill = max(0, $columns - 4 - AnsiUtils::visibleWidth($title) - AnsiUtils::visibleWidth($label)); + + $result = [ + self::color($borderColor, '╭─') + . $title + . self::color($borderColor, str_repeat('─', $fill)) + . $label + . self::color($borderColor, '─╮'), + ]; + + $side = self::color($borderColor, '│'); + + for ($i = 0; $i < $innerRows; $i++) { + $line = $lines[$i] ?? ''; + $result[] = $side . ' ' . self::fit($line, $innerWidth) . "\e[0m " . $side; + } + + $result[] = self::color($borderColor, '╰' . str_repeat('─', $columns - 2) . '╯'); + + return $result; + } + + /** + * Places the box lines centered on top of the given background lines. + * + * @param list $background + * @param list $box + * + * @return list + */ + public static function overlay(array $background, array $box, int $columns): array + { + $boxRows = count($box); + $boxWidth = $boxRows > 0 ? AnsiUtils::visibleWidth($box[0]) : 0; + + if ($boxRows === 0 || $boxRows > count($background) || $boxWidth > $columns) { + return $background; + } + + $top = intdiv(count($background) - $boxRows, 2); + $left = intdiv($columns - $boxWidth, 2); + + foreach ($box as $i => $boxLine) { + $line = self::fit($background[$top + $i], $columns); + + $composed = AnsiUtils::sliceByColumn($line, 0, $left, true) + . "\e[0m" . $boxLine . "\e[0m" + . AnsiUtils::sliceByColumn($line, $left + $boxWidth, $columns - $left - $boxWidth, true); + + // never exceed the available width, the renderer rejects such lines + $background[$top + $i] = AnsiUtils::truncateToWidth($composed, $columns, ''); + } + + return array_values($background); + } +} diff --git a/src/Console/Tui/Throughput.php b/src/Console/Tui/Throughput.php new file mode 100644 index 000000000..c86e5669f --- /dev/null +++ b/src/Console/Tui/Throughput.php @@ -0,0 +1,83 @@ +> */ + private array $samples = []; + + public function record(string $key, int|null $position, DateTimeImmutable $now): void + { + if ($position === null) { + unset($this->samples[$key]); + + return; + } + + $time = (float)$now->format('U.u'); + $samples = $this->samples[$key] ?? []; + $last = $samples[count($samples) - 1] ?? null; + + // the position went back, e.g. after a rebuild, so the old samples are meaningless + if ($last !== null && $position < $last[1]) { + $samples = []; + } + + $samples[] = [$time, $position]; + + // keep one sample older than the window as starting point + while (count($samples) > 2 && $samples[1][0] <= $time - self::WINDOW_SECONDS) { + array_shift($samples); + } + + $this->samples[$key] = $samples; + } + + /** @param list $keys */ + public function retain(array $keys): void + { + foreach ($this->samples as $key => $samples) { + if (in_array((string)$key, $keys, true)) { + continue; + } + + unset($this->samples[$key]); + } + } + + public function rate(string $key): float|null + { + $samples = $this->samples[$key] ?? []; + + if (count($samples) < 2) { + return null; + } + + [$startTime, $startPosition] = $samples[0]; + [$endTime, $endPosition] = $samples[count($samples) - 1]; + + if ($endTime <= $startTime) { + return null; + } + + return ($endPosition - $startPosition) / ($endTime - $startTime); + } +} diff --git a/src/Console/Tui/View/SubscriptionTableView.php b/src/Console/Tui/View/SubscriptionTableView.php new file mode 100644 index 000000000..d3763b5f9 --- /dev/null +++ b/src/Console/Tui/View/SubscriptionTableView.php @@ -0,0 +1,425 @@ + */ + private array $subscriptions = []; + + /** @var list */ + private array $visible = []; + + private int|null $head = null; + private DateTimeImmutable $now; + private readonly Throughput $throughput; + private string $filter = ''; + private string|null $selectedId = null; + private int $offset = 0; + private int $pageSize = 1; + + /** @var array */ + private array $marked = []; + + public function __construct() + { + $this->now = new DateTimeImmutable(); + $this->throughput = new Throughput(); + } + + /** @param list $subscriptions */ + public function update(array $subscriptions, int|null $head, DateTimeImmutable $now): void + { + usort($subscriptions, static fn (Subscription $a, Subscription $b) => strcmp($a->id(), $b->id())); + + $this->subscriptions = $subscriptions; + $this->head = $head; + $this->now = $now; + + $ids = array_map(static fn (Subscription $subscription) => $subscription->id(), $subscriptions); + + foreach ($subscriptions as $subscription) { + $this->throughput->record($subscription->id(), $subscription->position(), $now); + } + + $this->throughput->retain($ids); + + foreach (array_keys($this->marked) as $id) { + if (in_array($id, $ids, true)) { + continue; + } + + unset($this->marked[$id]); + } + + $this->applyFilter(); + } + + public function filter(): string + { + return $this->filter; + } + + public function changeFilter(string $filter): void + { + $this->filter = $filter; + $this->applyFilter(); + } + + /** @return list */ + public function subscriptions(): array + { + return $this->subscriptions; + } + + /** Processed messages per second, null as long as there is not enough data. */ + public function rate(string $id): float|null + { + return $this->throughput->rate($id); + } + + public function selected(): Subscription|null + { + foreach ($this->visible as $subscription) { + if ($subscription->id() === $this->selectedId) { + return $subscription; + } + } + + return null; + } + + public function toggleMark(): void + { + $selected = $this->selected(); + + if ($selected === null) { + return; + } + + if (isset($this->marked[$selected->id()])) { + unset($this->marked[$selected->id()]); + } else { + $this->marked[$selected->id()] = true; + } + + $this->moveBy(1); + } + + public function hasMarks(): bool + { + return $this->marked !== []; + } + + public function clearMarks(): void + { + $this->marked = []; + } + + /** + * The subscriptions an action applies to: the marked ones, or the selected one if nothing is marked. + * + * @return list + */ + public function targets(): array + { + if ($this->marked === []) { + $selected = $this->selected(); + + return $selected === null ? [] : [$selected]; + } + + return array_values(array_filter( + $this->subscriptions, + fn (Subscription $subscription) => isset($this->marked[$subscription->id()]), + )); + } + + public function title(): string + { + return Theme::bold(Theme::ACCENT, 'Subscriptions'); + } + + public function label(): string + { + $parts = []; + + if ($this->filter !== '') { + $parts[] = Theme::color(Theme::HIGHLIGHT, '/' . $this->filter); + } + + if ($this->marked !== []) { + $parts[] = Theme::color(Theme::HIGHLIGHT, sprintf('◆ %d marked', count($this->marked))); + } + + $parts[] = Theme::color(Theme::MUTED, count($this->visible) === count($this->subscriptions) + ? (string)count($this->subscriptions) + : sprintf('%d of %d', count($this->visible), count($this->subscriptions))); + + return implode(Theme::color(Theme::SUBTLE, ' · '), $parts); + } + + /** @return list */ + public function render(int $columns, int $rows): array + { + $this->pageSize = max(1, $rows - 2); + + $cells = array_map($this->cells(...), $this->visible); + $widths = $this->widths($cells, $columns - 2); + + $lines = [ + ' ' . Theme::color(Theme::MUTED, $this->row(self::COLUMNS, $widths)), + '', + ]; + + if ($this->visible === []) { + $lines[] = ' ' . Theme::color(Theme::MUTED, $this->subscriptions === [] + ? 'No subscriptions found' + : sprintf('No subscriptions match "%s"', $this->filter)); + + return $lines; + } + + $selectedIndex = $this->selectedIndex(); + + if ($selectedIndex < $this->offset) { + $this->offset = $selectedIndex; + } elseif ($selectedIndex >= $this->offset + $this->pageSize) { + $this->offset = $selectedIndex - $this->pageSize + 1; + } + + $this->offset = max(0, min($this->offset, count($this->visible) - $this->pageSize)); + + foreach (array_slice($this->visible, $this->offset, $this->pageSize, true) as $index => $subscription) { + $selected = $index === $selectedIndex; + $marked = isset($this->marked[$subscription->id()]); + + $gutter = ($selected ? Theme::color(Theme::ACCENT, '▌') : ' ') + . ($marked ? Theme::color(Theme::HIGHLIGHT, '◆') : ' '); + + $row = $this->row($cells[$index], $widths); + + if (!$selected) { + $lines[] = $gutter . $row; + + continue; + } + + $background = "\e[" . Theme::SELECTED_BACKGROUND . 'm'; + $lines[] = $gutter . $background + . AnsiUtils::reapplyBackgroundAfterResets(Theme::fit($row, $columns - 2), $background) + . "\e[0m"; + } + + return $lines; + } + + public function moveBy(int $delta): void + { + if ($this->visible === []) { + return; + } + + $index = max(0, min(count($this->visible) - 1, $this->selectedIndex() + $delta)); + $this->selectedId = $this->visible[$index]->id(); + } + + public function pageUp(): void + { + $this->moveBy(-$this->pageSize); + } + + public function pageDown(): void + { + $this->moveBy($this->pageSize); + } + + public function moveToStart(): void + { + $this->moveBy(-count($this->visible)); + } + + public function moveToEnd(): void + { + $this->moveBy(count($this->visible)); + } + + private function applyFilter(): void + { + $needle = strtolower($this->filter); + + $this->visible = array_values(array_filter( + $this->subscriptions, + static fn (Subscription $subscription) => $needle === '' + || str_contains(strtolower($subscription->id()), $needle) + || str_contains(strtolower($subscription->group()), $needle) + || str_contains($subscription->status()->value, $needle) + || str_contains($subscription->runMode()->value, $needle), + )); + + if ($this->selected() !== null) { + return; + } + + $this->selectedId = isset($this->visible[0]) ? $this->visible[0]->id() : null; + $this->offset = 0; + } + + private function selectedIndex(): int + { + foreach ($this->visible as $index => $subscription) { + if ($subscription->id() === $this->selectedId) { + return $index; + } + } + + return 0; + } + + /** @return list */ + private function cells(Subscription $subscription): array + { + $position = $subscription->position(); + $lag = SubscriptionFormatter::lag($subscription, $this->head); + $error = $subscription->subscriptionError()?->errorMessage; + $rate = $this->throughput->rate($subscription->id()); + + return [ + Theme::color(Theme::TEXT, $subscription->id()), + Theme::color(Theme::MUTED, $subscription->group()), + Theme::color(Theme::MUTED, SubscriptionFormatter::runMode($subscription)), + $position === null || $this->head === null + ? Theme::color(Theme::SUBTLE, '–') + : Theme::progressBar($position, $this->head, self::PROGRESS_BAR_WIDTH) + . ' ' . Theme::fitRight(Theme::color(Theme::MUTED, Theme::percent($position, $this->head)), 4), + $position === null ? Theme::color(Theme::SUBTLE, '–') : Theme::color(Theme::TEXT, Theme::number($position)), + match (true) { + $lag === null => Theme::color(Theme::SUBTLE, '–'), + $lag === 0 => Theme::color(Theme::SUBTLE, '0'), + default => Theme::color(Theme::WARNING, Theme::number($lag)), + }, + match (true) { + $rate === null => Theme::color(Theme::SUBTLE, '–'), + $rate < 0.05 => Theme::color(Theme::SUBTLE, '0'), + default => Theme::color(Theme::SUCCESS, Theme::rate($rate)), + }, + Theme::status($subscription->status()), + $subscription->retryAttempt() === 0 + ? Theme::color(Theme::SUBTLE, '0') + : Theme::color(Theme::WARNING, (string)$subscription->retryAttempt()), + Theme::color(Theme::MUTED, SubscriptionFormatter::age($subscription->lastSavedAt(), $this->now)), + $error === null ? '' : Theme::color('38;5;174', $error), + ]; + } + + /** + * Calculates the column widths, dropping optional columns on small terminals. + * A width of 0 hides the column. + * + * @param list> $cells + * + * @return list + */ + private function widths(array $cells, int $columns): array + { + $widths = []; + + foreach (self::COLUMNS as $index => $name) { + $width = AnsiUtils::visibleWidth($name); + + foreach ($cells as $row) { + $width = max($width, AnsiUtils::visibleWidth($row[$index])); + } + + $widths[] = $width; + } + + $errorIndex = count($widths) - 1; + $widths[$errorIndex] = 0; + + foreach (self::OPTIONAL as $optional) { + if ($this->totalWidth($widths) <= $columns) { + break; + } + + $widths[(int)array_keys(self::COLUMNS, $optional, true)[0]] = 0; + } + + $remaining = $columns - $this->totalWidth($widths) - 2; + + if ($remaining >= self::MIN_ERROR_WIDTH) { + $widths[$errorIndex] = $remaining; + } + + return $widths; + } + + /** @param list $widths */ + private function totalWidth(array $widths): int + { + $visible = array_filter($widths, static fn (int $width) => $width > 0); + + return array_sum($visible) + 2 * max(0, count($visible) - 1); + } + + /** + * @param list $cells + * @param list $widths + */ + private function row(array $cells, array $widths): string + { + $parts = []; + + foreach ($cells as $index => $cell) { + $width = $widths[$index] ?? 0; + + if ($width === 0) { + continue; + } + + $parts[] = in_array(self::COLUMNS[$index], self::RIGHT_ALIGNED, true) + ? Theme::fitRight($cell, $width) + : Theme::fit($cell, $width); + } + + return implode(self::GAP, $parts); + } +} diff --git a/src/Console/Tui/View/TextView.php b/src/Console/Tui/View/TextView.php new file mode 100644 index 000000000..b5fe6152b --- /dev/null +++ b/src/Console/Tui/View/TextView.php @@ -0,0 +1,97 @@ + $lines */ + public function __construct( + private string $title, + private array $lines = [], + ) { + } + + /** @param list $lines */ + public function update(string $title, array $lines): void + { + $this->title = $title; + $this->lines = $lines; + $this->offset = min($this->offset, $this->maxOffset()); + } + + public function title(): string + { + return Theme::bold(Theme::ACCENT, $this->title); + } + + public function label(): string + { + if (count($this->lines) <= $this->pageSize) { + return ''; + } + + return Theme::color(Theme::MUTED, sprintf( + '%d–%d of %d', + $this->offset + 1, + min(count($this->lines), $this->offset + $this->pageSize), + count($this->lines), + )); + } + + /** @return list */ + public function render(int $columns, int $rows): array + { + $this->pageSize = max(1, $rows); + $this->offset = min($this->offset, $this->maxOffset()); + + return array_slice($this->lines, $this->offset, $this->pageSize); + } + + public function moveBy(int $delta): void + { + $this->offset = max(0, min($this->maxOffset(), $this->offset + $delta)); + } + + public function pageUp(): void + { + $this->moveBy(-$this->pageSize); + } + + public function pageDown(): void + { + $this->moveBy($this->pageSize); + } + + public function moveToStart(): void + { + $this->offset = 0; + } + + public function moveToEnd(): void + { + $this->offset = $this->maxOffset(); + } + + private function maxOffset(): int + { + return max(0, count($this->lines) - $this->pageSize); + } +} diff --git a/src/Console/Tui/View/View.php b/src/Console/Tui/View/View.php new file mode 100644 index 000000000..2eccd404b --- /dev/null +++ b/src/Console/Tui/View/View.php @@ -0,0 +1,30 @@ + */ + public function render(int $columns, int $rows): array; + + public function moveBy(int $delta): void; + + public function pageUp(): void; + + public function pageDown(): void; + + public function moveToStart(): void; + + public function moveToEnd(): void; +} diff --git a/src/Console/Tui/Widget/FooterWidget.php b/src/Console/Tui/Widget/FooterWidget.php new file mode 100644 index 000000000..1f815b736 --- /dev/null +++ b/src/Console/Tui/Widget/FooterWidget.php @@ -0,0 +1,100 @@ + */ + private array $hints = []; + private string|null $prompt = null; + private string $message = ''; + private string $messageColor = Theme::MUTED; + + /** @param list $hints list of [key, label] */ + public function setHints(array $hints): void + { + $this->hints = $hints; + $this->invalidate(); + } + + public function setPrompt(string|null $prompt): void + { + $this->prompt = $prompt; + $this->invalidate(); + } + + public function setMessage(string $message, string $color = Theme::MUTED): void + { + $this->message = $message; + $this->messageColor = $color; + $this->invalidate(); + } + + /** @return list */ + public function render(RenderContext $context): array + { + $columns = $context->getColumns(); + + $right = ''; + + if ($this->message !== '') { + $icon = match ($this->messageColor) { + Theme::SUCCESS => '✓', + Theme::WARNING => '!', + Theme::DANGER => '✗', + default => '', + }; + + $right = Theme::color($this->messageColor, ($icon === '' ? '' : $icon . ' ') . $this->message) . ' '; + } + + if ($this->prompt !== null) { + $prompt = ' ' . Theme::bold(Theme::HIGHLIGHT, '/') . ' ' + . Theme::color(Theme::TEXT, $this->prompt) + . Theme::color(Theme::HIGHLIGHT, '▏'); + + return [ + Theme::fit($prompt, $columns), + Theme::spread(' ' . Theme::keyHint('enter', 'apply') . ' ' . Theme::keyHint('esc', 'clear'), $right, $columns), + ]; + } + + $lines = ['', '']; + $line = 0; + + foreach ($this->hints as [$key, $label]) { + $hint = ' ' . Theme::keyHint($key, $label) . ' '; + $available = $line === 1 ? $columns - AnsiUtils::visibleWidth($right) - 1 : $columns; + + if (AnsiUtils::visibleWidth($lines[$line] . $hint) > max(0, $available)) { + // hints that do not fit are left out instead of being cut in half + if ($line === 1) { + break; + } + + $line = 1; + } + + $lines[$line] .= $hint; + } + + return [ + Theme::fit($lines[0], $columns), + Theme::spread($lines[1], $right, $columns), + ]; + } +} diff --git a/src/Console/Tui/Widget/HeaderWidget.php b/src/Console/Tui/Widget/HeaderWidget.php new file mode 100644 index 000000000..a0129648e --- /dev/null +++ b/src/Console/Tui/Widget/HeaderWidget.php @@ -0,0 +1,69 @@ + */ + private array $crumbs = []; + + /** @var list */ + private array $meta = []; + + private string $summary = ''; + + /** @param list $crumbs */ + public function setCrumbs(array $crumbs): void + { + $this->crumbs = $crumbs; + $this->invalidate(); + } + + /** @param list $meta */ + public function setMeta(array $meta): void + { + $this->meta = $meta; + $this->invalidate(); + } + + public function setSummary(string $summary): void + { + $this->summary = $summary; + $this->invalidate(); + } + + /** @return list */ + public function render(RenderContext $context): array + { + $columns = $context->getColumns(); + + $crumbs = array_map( + static fn (string $crumb) => Theme::color(Theme::TEXT, $crumb), + $this->crumbs, + ); + + $left = ' ' . Theme::bold(Theme::ACCENT, '◆ event-sourcing') + . ($crumbs === [] ? '' : Theme::color(Theme::SUBTLE, ' › ') . implode(Theme::color(Theme::SUBTLE, ' › '), $crumbs)); + + $right = implode(Theme::color(Theme::SUBTLE, ' · '), $this->meta) . ' '; + + return [ + Theme::spread($left, $right, $columns), + Theme::fit(' ' . $this->summary, $columns), + ]; + } +} diff --git a/src/Console/Tui/Widget/PanelWidget.php b/src/Console/Tui/Widget/PanelWidget.php new file mode 100644 index 000000000..1785b4652 --- /dev/null +++ b/src/Console/Tui/Widget/PanelWidget.php @@ -0,0 +1,110 @@ +|null */ + private array|null $dialog = null; + private string $dialogTitle = ''; + private string $dialogColor = Theme::ACCENT; + + public function __construct( + private View $view, + ) { + } + + public function view(): View + { + return $this->view; + } + + public function show(View $view): void + { + $this->view = $view; + $this->invalidate(); + } + + /** @param list $lines */ + public function openDialog(string $title, array $lines, string $color = Theme::ACCENT): void + { + $this->dialogTitle = $title; + $this->dialogColor = $color; + $this->dialog = $lines; + $this->invalidate(); + } + + public function closeDialog(): void + { + $this->dialog = null; + $this->invalidate(); + } + + public function hasDialog(): bool + { + return $this->dialog !== null; + } + + public function expandVertically(bool $expand): static + { + return $this; + } + + public function isVerticallyExpanded(): bool + { + return true; + } + + /** @return list */ + public function render(RenderContext $context): array + { + $columns = $context->getColumns(); + $rows = max(3, $context->getRows()); + + $content = $this->view->render(max(1, $columns - 4), $rows - 2); + $lines = Theme::frame($this->view->title(), $content, $columns, $rows, $this->view->label()); + + if ($this->dialog === null) { + return $lines; + } + + $width = 0; + + foreach ($this->dialog as $line) { + $width = max($width, AnsiUtils::visibleWidth($line)); + } + + $width = min($columns - 4, max(44, $width + 8)); + $content = ['', ...array_map(static fn (string $line) => ' ' . $line, $this->dialog), '']; + $box = Theme::frame( + Theme::bold($this->dialogColor, $this->dialogTitle), + $content, + $width, + count($content) + 2, + '', + $this->dialogColor, + ); + + return Theme::overlay($lines, $box, $columns); + } +} diff --git a/tests/Unit/Console/Command/SubscriptionDashboardCommandTest.php b/tests/Unit/Console/Command/SubscriptionDashboardCommandTest.php new file mode 100644 index 000000000..e5faae2d9 --- /dev/null +++ b/tests/Unit/Console/Command/SubscriptionDashboardCommandTest.php @@ -0,0 +1,52 @@ +execute([], ['interactive' => false]); + + self::assertSame(2, $commandTester->getStatusCode()); + self::assertStringContainsString('requires an interactive terminal', $commandTester->getDisplay()); + } + + public function testRunWithFilter(): void + { + $engine = new RecordingSubscriptionEngine([new Subscription('profile')]); + $runner = new DashboardRunner(); + + $commandTester = new CommandTester(new SubscriptionDashboardCommand($engine, terminal: $runner->terminal)); + $screen = $runner->run(static fn () => $commandTester->execute(['--group' => ['projector']])); + + self::assertSame(0, $commandTester->getStatusCode()); + self::assertEquals(new SubscriptionEngineCriteria(null, ['projector']), $engine->criteria[0]); + self::assertStringContainsString('group=projector', $screen); + } +} diff --git a/tests/Unit/Console/Command/SubscriptionRunCommandTest.php b/tests/Unit/Console/Command/SubscriptionRunCommandTest.php index 0dc1f83ac..ce9f27340 100644 --- a/tests/Unit/Console/Command/SubscriptionRunCommandTest.php +++ b/tests/Unit/Console/Command/SubscriptionRunCommandTest.php @@ -52,6 +52,29 @@ public function testRun(): void self::assertSame(0, $commandTester->getStatusCode()); } + public function testRunStopsWhenFinished(): void + { + $store = $this->createMock(Store::class); + + $engine = $this->createMock(SubscriptionEngine::class); + $engine + ->expects($this->once()) + ->method('subscriptions') + ->willReturn([new Subscription('foo')]); + $engine + ->expects($this->exactly(2)) + ->method('execute') + ->willReturnOnConsecutiveCalls( + new ProcessedResult(100, false), + new ProcessedResult(20, true), + ); + + $commandTester = new CommandTester(new SubscriptionRunCommand($engine, $store)); + $commandTester->execute(['--stop-when-finished' => true, '--sleep' => 0]); + + self::assertSame(0, $commandTester->getStatusCode()); + } + public function testRunWithRebuild(): void { $store = $this->createMock(Store::class); diff --git a/tests/Unit/Console/Tui/DashboardRunner.php b/tests/Unit/Console/Tui/DashboardRunner.php new file mode 100644 index 000000000..e7a938f82 --- /dev/null +++ b/tests/Unit/Console/Tui/DashboardRunner.php @@ -0,0 +1,56 @@ +terminal = new VirtualTerminal($columns, $rows); + } + + /** + * @param Closure(): mixed $run starts the (blocking) dashboard + * @param list $inputs + */ + public function run(Closure $run, array $inputs = []): string + { + $delay = self::STEP; + + foreach ($inputs as $input) { + EventLoop::delay($delay, fn () => $this->terminal->simulateInput($input)); + $delay += self::STEP; + } + + EventLoop::delay($delay + self::STEP, function (): void { + $buffer = new ScreenBuffer($this->columns, $this->rows); + $buffer->write($this->terminal->getOutput()); + + $this->screen = $buffer->getScreen(); + + $this->terminal->simulateInput("\x03"); + }); + + $run(); + + return $this->screen; + } +} diff --git a/tests/Unit/Console/Tui/ProcessActionRunnerTest.php b/tests/Unit/Console/Tui/ProcessActionRunnerTest.php new file mode 100644 index 000000000..6ef563295 --- /dev/null +++ b/tests/Unit/Console/Tui/ProcessActionRunnerTest.php @@ -0,0 +1,69 @@ +runAction($runner, SubscriptionAction::Rebuild, ['profile'])); + self::assertSame( + "event-sourcing:subscription:remove --force --id=profile --no-interaction\n" + . "event-sourcing:subscription:boot --setup --message-limit=50 --id=profile --no-interaction\n", + file_get_contents($log), + ); + + unlink($log); + } + + public function testFailedCommand(): void + { + $runner = new ProcessActionRunner([PHP_BINARY, '-r', 'echo "Something went wrong\n"; exit(3);', '--'], null, 0.01); + + self::assertSame( + ['"event-sourcing:subscription:pause" exited with code 3: Something went wrong'], + $this->runAction($runner, SubscriptionAction::Pause, ['profile']), + ); + } + + /** + * @param list $ids + * + * @return list + */ + private function runAction(ProcessActionRunner $runner, SubscriptionAction $action, array $ids): array + { + /** @var Suspension> $suspension */ + $suspension = EventLoop::getSuspension(); + $runner->run($action, $ids, static fn (array $errors) => $suspension->resume($errors)); + + return $suspension->suspend(); + } +} diff --git a/tests/Unit/Console/Tui/RecordingSubscriptionEngine.php b/tests/Unit/Console/Tui/RecordingSubscriptionEngine.php new file mode 100644 index 000000000..662a328e0 --- /dev/null +++ b/tests/Unit/Console/Tui/RecordingSubscriptionEngine.php @@ -0,0 +1,42 @@ + */ + public array $commands = []; + + /** @var list */ + public array $criteria = []; + + /** @param list $subscriptions */ + public function __construct( + public array $subscriptions = [], + public Result $result = new Result(), + ) { + } + + public function execute(Command $command): Result + { + $this->commands[] = $command; + + return $this->result; + } + + /** @return list */ + public function subscriptions(SubscriptionEngineCriteria|null $criteria = null): array + { + $this->criteria[] = $criteria; + + return $this->subscriptions; + } +} diff --git a/tests/Unit/Console/Tui/SubscriptionDashboardTest.php b/tests/Unit/Console/Tui/SubscriptionDashboardTest.php new file mode 100644 index 000000000..9e27c672c --- /dev/null +++ b/tests/Unit/Console/Tui/SubscriptionDashboardTest.php @@ -0,0 +1,159 @@ +runDashboard($engine, []); + + self::assertStringContainsString('profile', $screen); + self::assertStringContainsString('welcome_mail', $screen); + self::assertStringContainsString('● active', $screen); + self::assertStringContainsString('‖ paused', $screen); + self::assertStringContainsString('head 3', $screen); + } + + public function testPauseSelectedSubscription(): void + { + $engine = new RecordingSubscriptionEngine([ + new Subscription('a', status: Status::Active), + new Subscription('b', status: Status::Active), + ]); + + $screen = $this->runDashboard($engine, [self::DOWN, 'p']); + + self::assertEquals([new Pause(['b'])], $engine->commands); + self::assertStringContainsString('Pause "b" done', $screen); + } + + public function testActionNotPossibleForStatus(): void + { + $engine = new RecordingSubscriptionEngine([new Subscription('a', status: Status::New)]); + + $screen = $this->runDashboard($engine, ['p']); + + self::assertSame([], $engine->commands); + self::assertStringContainsString('Pause is not possible for subscription "a" with status "new"', $screen); + } + + public function testRemoveNeedsConfirmation(): void + { + $engine = new RecordingSubscriptionEngine([new Subscription('a', status: Status::Active)]); + + $screen = $this->runDashboard($engine, [self::CTRL_D]); + + self::assertSame([], $engine->commands); + self::assertStringContainsString('Remove "a"?', $screen); + + $this->runDashboard($engine, [self::CTRL_D, 'y']); + + self::assertEquals([new Remove(['a'])], $engine->commands); + } + + public function testFilter(): void + { + $engine = new RecordingSubscriptionEngine([ + new Subscription('profile', status: Status::Active), + new Subscription('welcome_mail'), + ]); + + // typed keys must not trigger actions like "p" (pause) while filtering + $screen = $this->runDashboard($engine, ['/', 'm', 'a', 'i', 'l', 'p', "\x7f", self::ENTER]); + + self::assertSame([], $engine->commands); + self::assertStringContainsString('welcome_mail', $screen); + self::assertStringNotContainsString('profile', $screen); + } + + public function testDescribe(): void + { + $engine = new RecordingSubscriptionEngine([ + new Subscription( + 'profile', + 'projector', + RunMode::FromBeginning, + Status::Error, + 2, + SubscriptionError::fromThrowable(Status::Active, new RuntimeException('projection broken')), + ), + ]); + + $screen = $this->runDashboard($engine, [self::ENTER]); + + self::assertStringContainsString('Overview', $screen); + self::assertStringContainsString('was active', $screen); + self::assertStringContainsString('RuntimeException', $screen); + self::assertStringContainsString('projection broken', $screen); + } + + public function testShowLoadingError(): void + { + $engine = $this->createStub(SubscriptionEngine::class); + $engine->method('subscriptions')->willThrowException(new RuntimeException('connection refused')); + + $runner = new DashboardRunner(); + $screen = $runner->run( + static fn () => (new SubscriptionDashboard($engine, terminal: $runner->terminal))->run(), + ); + + self::assertStringContainsString('Failed to load subscriptions: connection refused', $screen); + } + + /** @param list $inputs */ + private function runDashboard(RecordingSubscriptionEngine $engine, array $inputs): string + { + $store = new InMemoryStore([ + Message::create(new stdClass()), + Message::create(new stdClass()), + Message::create(new stdClass()), + ]); + + $runner = new DashboardRunner(); + + return $runner->run( + static fn () => (new SubscriptionDashboard($engine, $store, terminal: $runner->terminal))->run(), + $inputs, + ); + } +} diff --git a/tests/Unit/Console/Tui/ThroughputTest.php b/tests/Unit/Console/Tui/ThroughputTest.php new file mode 100644 index 000000000..8bc812d44 --- /dev/null +++ b/tests/Unit/Console/Tui/ThroughputTest.php @@ -0,0 +1,59 @@ +record('profile', 100, new DateTimeImmutable('@1000')); + self::assertNull($throughput->rate('profile')); + + $throughput->record('profile', 300, new DateTimeImmutable('@1002')); + $throughput->record('profile', 500, new DateTimeImmutable('@1004')); + self::assertSame(100.0, $throughput->rate('profile')); + + // only the last 10 seconds count + $throughput->record('profile', 700, new DateTimeImmutable('@1014')); + self::assertSame(20.0, $throughput->rate('profile')); + } + + public function testResetWhenPositionGoesBack(): void + { + $throughput = new Throughput(); + + $throughput->record('profile', 500, new DateTimeImmutable('@1000')); + $throughput->record('profile', 0, new DateTimeImmutable('@1002')); + self::assertNull($throughput->rate('profile')); + + $throughput->record('profile', 10, new DateTimeImmutable('@1004')); + self::assertSame(5.0, $throughput->rate('profile')); + } + + public function testForget(): void + { + $throughput = new Throughput(); + + $throughput->record('profile', 100, new DateTimeImmutable('@1000')); + $throughput->record('profile', 200, new DateTimeImmutable('@1001')); + $throughput->record('123', 100, new DateTimeImmutable('@1000')); + $throughput->record('123', 200, new DateTimeImmutable('@1001')); + + $throughput->retain(['123']); + self::assertNull($throughput->rate('profile')); + self::assertSame(100.0, $throughput->rate('123')); + + $throughput->record('123', null, new DateTimeImmutable('@1002')); + self::assertNull($throughput->rate('123')); + } +}