From 411c0f697f13f885be4024a870c9260f8f0bd895 Mon Sep 17 00:00:00 2001 From: David Badura Date: Wed, 7 Oct 2026 09:59:57 +0200 Subject: [PATCH 1/4] Add interactive subscription dashboard A k9s like terminal UI built on symfony/tui to monitor and manage subscriptions. It shows status, progress and lag in a live updating table, has a describe view with the error and stack trace, and runs the subscription commands (setup, boot, run, pause, reactivate, refresh, teardown, remove, rebuild) on the selected or marked subscriptions. symfony/tui requires PHP 8.4, so it is only a dev dependency and a suggestion. The unit workflow removes it for PHP 8.2/8.3 and lowest dependencies, and the tests are skipped when it is missing. bin/subscription-dashboard-demo starts the dashboard with an in-memory store and a few example subscribers. --- .github/workflows/unit.yml | 5 + bin/subscription-dashboard-demo | 229 ++++++ composer.json | 4 +- composer.lock | 152 +++- docs/cli.md | 81 +++ docs/subscription.md | 1 + .../Command/SubscriptionDashboardCommand.php | 93 +++ src/Console/Tui/SubscriptionAction.php | 134 ++++ src/Console/Tui/SubscriptionDashboard.php | 687 ++++++++++++++++++ src/Console/Tui/SubscriptionFormatter.php | 222 ++++++ src/Console/Tui/Theme.php | 246 +++++++ .../Tui/View/SubscriptionTableView.php | 404 ++++++++++ src/Console/Tui/View/TextView.php | 97 +++ src/Console/Tui/View/View.php | 30 + src/Console/Tui/Widget/FooterWidget.php | 100 +++ src/Console/Tui/Widget/HeaderWidget.php | 69 ++ src/Console/Tui/Widget/PanelWidget.php | 110 +++ .../SubscriptionDashboardCommandTest.php | 52 ++ tests/Unit/Console/Tui/DashboardRunner.php | 56 ++ .../Tui/RecordingSubscriptionEngine.php | 42 ++ .../Console/Tui/SubscriptionDashboardTest.php | 159 ++++ 21 files changed, 2971 insertions(+), 2 deletions(-) create mode 100755 bin/subscription-dashboard-demo create mode 100644 src/Console/Command/SubscriptionDashboardCommand.php create mode 100644 src/Console/Tui/SubscriptionAction.php create mode 100644 src/Console/Tui/SubscriptionDashboard.php create mode 100644 src/Console/Tui/SubscriptionFormatter.php create mode 100644 src/Console/Tui/Theme.php create mode 100644 src/Console/Tui/View/SubscriptionTableView.php create mode 100644 src/Console/Tui/View/TextView.php create mode 100644 src/Console/Tui/View/View.php create mode 100644 src/Console/Tui/Widget/FooterWidget.php create mode 100644 src/Console/Tui/Widget/HeaderWidget.php create mode 100644 src/Console/Tui/Widget/PanelWidget.php create mode 100644 tests/Unit/Console/Command/SubscriptionDashboardCommandTest.php create mode 100644 tests/Unit/Console/Tui/DashboardRunner.php create mode 100644 tests/Unit/Console/Tui/RecordingSubscriptionEngine.php create mode 100644 tests/Unit/Console/Tui/SubscriptionDashboardTest.php 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..7047147fb 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 @@ -330,6 +331,86 @@ 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 and lag, +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 +``` +### 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 | + +:::warning +Run, boot and rebuild are executed in the dashboard process and block the UI until they are done. +Use the `--message-limit` option to keep them short and the [run command](#subscription-commands) as worker. +::: + +:::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..2b76cc414 --- /dev/null +++ b/src/Console/Command/SubscriptionDashboardCommand.php @@ -0,0 +1,93 @@ +addOption( + 'refresh', + null, + InputOption::VALUE_REQUIRED, + 'How often the subscriptions are reloaded in seconds', + 2, + ) + ->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, + InputHelper::nullablePositiveInt($input->getOption('message-limit')), + $this->terminal, + $this->clock, + ); + + $dashboard->run(); + + return 0; + } +} 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..6787e344c --- /dev/null +++ b/src/Console/Tui/SubscriptionDashboard.php @@ -0,0 +1,687 @@ + */ + private array $pendingTargets = []; + + private int|null $head = null; + private DateTimeImmutable|null $lastReload = null; + private int $flashUntil = 0; + + /** @param positive-int|null $messageLimit */ + 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, + private readonly int|null $messageLimit = null, + TerminalInterface|null $terminal = null, + private readonly ClockInterface|null $clock = null, + ) { + $this->keybindings = self::keybindings(); + $this->tui = new Tui(self::styleSheet(), $terminal ?? new Terminal(), $this->keybindings); + + $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); + $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->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 + { + $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); + + // show the progress before the (blocking) command is executed + $this->flash(sprintf('%s %s...', $action->label(), $this->describeTargets($targets)), Theme::WARNING); + $this->tui->requestRender(); + $this->tui->processRender(); + + $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(); + } + + $this->table->clearMarks(); + $this->reload(); + + if ($errors === []) { + $this->flash(sprintf('%s %s done', $action->label(), $this->describeTargets($targets)), Theme::SUCCESS); + + return; + } + + $this->flash( + sprintf('%s failed: %s', $action->label(), $errors[0]) . (count($errors) > 1 ? sprintf(' (+%d more)', count($errors) - 1) : ''), + Theme::DANGER, + ); + } + + /** @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()), + ); + } + + 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()); + } + + $meta[] = Theme::color(Theme::MUTED, 'head ') . Theme::color(Theme::TEXT, $this->head === null ? '–' : Theme::number($this->head)); + $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..b0dc80865 --- /dev/null +++ b/src/Console/Tui/SubscriptionFormatter.php @@ -0,0 +1,222 @@ +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 '-'; + } + + $seconds = max(0, $now->getTimestamp() - $time->getTimestamp()); + + 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): 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('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))) : ''); + } + + /** @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..77fab393c --- /dev/null +++ b/src/Console/Tui/Theme.php @@ -0,0 +1,246 @@ + '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, '.', ','); + } + + 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/View/SubscriptionTableView.php b/src/Console/Tui/View/SubscriptionTableView.php new file mode 100644 index 000000000..2b07340e4 --- /dev/null +++ b/src/Console/Tui/View/SubscriptionTableView.php @@ -0,0 +1,404 @@ + */ + private array $subscriptions = []; + + /** @var list */ + private array $visible = []; + + private int|null $head = null; + private DateTimeImmutable $now; + 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(); + } + + /** @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 (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; + } + + 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; + + 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)), + }, + 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..7fe7491e5 --- /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 . ' ' . $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/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/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, + ); + } +} From 9f06c67957e5ca0fb69bc2d94608ba7cc1c71b36 Mon Sep 17 00:00:00 2001 From: David Badura Date: Wed, 7 Oct 2026 10:22:29 +0200 Subject: [PATCH 2/4] Run dashboard actions in a separate process The actions are now executed with the existing subscription commands in a separate process, so the dashboard keeps rendering and reacting to input while e.g. a long boot or rebuild is running. A spinner shows the running action, and a second action has to wait until it is done. The process is polled from the event loop and its output goes into a temporary file, so this also works on windows where revolt can not wait on pipes. If the subscription commands are not registered in the same console application (like in the demo), or with --in-process, the actions still run in the dashboard process. --- docs/cli.md | 14 +- .../Command/SubscriptionDashboardCommand.php | 48 ++++- src/Console/Tui/ActionRunner/ActionRunner.php | 21 +++ .../ActionRunner/InProcessActionRunner.php | 49 +++++ .../Tui/ActionRunner/ProcessActionRunner.php | 173 ++++++++++++++++++ src/Console/Tui/SubscriptionDashboard.php | 84 +++++++-- src/Console/Tui/Widget/FooterWidget.php | 4 +- .../Console/Tui/ProcessActionRunnerTest.php | 69 +++++++ 8 files changed, 434 insertions(+), 28 deletions(-) create mode 100644 src/Console/Tui/ActionRunner/ActionRunner.php create mode 100644 src/Console/Tui/ActionRunner/InProcessActionRunner.php create mode 100644 src/Console/Tui/ActionRunner/ProcessActionRunner.php create mode 100644 tests/Unit/Console/Tui/ProcessActionRunnerTest.php diff --git a/docs/cli.md b/docs/cli.md index 7047147fb..229a7898d 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -377,6 +377,15 @@ and `--message-limit` sets how many messages a run, boot or rebuild processes at ```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 processes up to `--message-limit` messages once, 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. @@ -401,11 +410,6 @@ for example `pause` only for active, booting or failing subscriptions. | `?` | Show all shortcuts | | `q` | Quit | -:::warning -Run, boot and rebuild are executed in the dashboard process and block the UI until they are done. -Use the `--message-limit` option to keep them short and the [run command](#subscription-commands) as worker. -::: - :::danger Remove, teardown and rebuild delete the data of the subscribers. The dashboard asks for a confirmation before it executes them. diff --git a/src/Console/Command/SubscriptionDashboardCommand.php b/src/Console/Command/SubscriptionDashboardCommand.php index 2b76cc414..ec5ddab03 100644 --- a/src/Console/Command/SubscriptionDashboardCommand.php +++ b/src/Console/Command/SubscriptionDashboardCommand.php @@ -6,6 +6,9 @@ use Patchlevel\EventSourcing\Console\InputHelper; use Patchlevel\EventSourcing\Console\OutputStyle; +use Patchlevel\EventSourcing\Console\Tui\ActionRunner\ActionRunner; +use Patchlevel\EventSourcing\Console\Tui\ActionRunner\InProcessActionRunner; +use Patchlevel\EventSourcing\Console\Tui\ActionRunner\ProcessActionRunner; use Patchlevel\EventSourcing\Console\Tui\SubscriptionDashboard; use Patchlevel\EventSourcing\Store\Store; use Patchlevel\EventSourcing\Subscription\Engine\SubscriptionEngine; @@ -17,10 +20,16 @@ use Symfony\Component\Tui\Terminal\TerminalInterface; use Symfony\Component\Tui\Tui; +use function array_filter; use function class_exists; +use function is_array; +use function is_file; use function is_numeric; +use function is_string; use function max; +use const PHP_BINARY; + /** @experimental */ #[AsCommand( 'event-sourcing:subscription:dashboard', @@ -49,6 +58,12 @@ protected function configure(): void '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, @@ -81,7 +96,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int $this->store, $this->subscriptionEngineCriteria($input), is_numeric($refresh) ? max(0.1, (float)$refresh) : 2.0, - InputHelper::nullablePositiveInt($input->getOption('message-limit')), + $this->actionRunner($input), $this->terminal, $this->clock, ); @@ -90,4 +105,35 @@ protected function execute(InputInterface $input, OutputInterface $output): int 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/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..d4fb6685b --- /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', '--run-limit=1', '--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/SubscriptionDashboard.php b/src/Console/Tui/SubscriptionDashboard.php index 6787e344c..28ecb3750 100644 --- a/src/Console/Tui/SubscriptionDashboard.php +++ b/src/Console/Tui/SubscriptionDashboard.php @@ -5,6 +5,8 @@ namespace Patchlevel\EventSourcing\Console\Tui; use DateTimeImmutable; +use Patchlevel\EventSourcing\Console\Tui\ActionRunner\ActionRunner; +use Patchlevel\EventSourcing\Console\Tui\ActionRunner\InProcessActionRunner; use Patchlevel\EventSourcing\Console\Tui\View\SubscriptionTableView; use Patchlevel\EventSourcing\Console\Tui\View\TextView; use Patchlevel\EventSourcing\Console\Tui\Widget\FooterWidget; @@ -53,7 +55,10 @@ final class SubscriptionDashboard { private const FLASH_SECONDS = 5; + private const SPINNER = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏']; + private readonly Tui $tui; + private readonly ActionRunner $runner; private readonly Keybindings $keybindings; private readonly HeaderWidget $header; @@ -75,16 +80,20 @@ final class SubscriptionDashboard private DateTimeImmutable|null $lastReload = null; private int $flashUntil = 0; - /** @param positive-int|null $messageLimit */ + 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, - private readonly int|null $messageLimit = null, + 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); @@ -119,6 +128,7 @@ public function run(): void $this->tui->run(); } finally { EventLoop::cancel($timer); + $this->stopSpinner(); $terminal->write("\x1b[?1049l"); } } @@ -314,6 +324,12 @@ private function back(): void 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 === []) { @@ -361,39 +377,67 @@ private function trigger(SubscriptionAction $action): void 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)); - // show the progress before the (blocking) command is executed - $this->flash(sprintf('%s %s...', $action->label(), $this->describeTargets($targets)), Theme::WARNING); + $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(); - $errors = []; + $this->runner->run($action, $ids, function (array $errors) use ($action, $label): void { + $this->running = null; + $this->stopSpinner(); + $this->table->clearMarks(); + $this->reload(); - 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); - } + 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, + ); } - } catch (Throwable $e) { - $errors[] = $e->getMessage(); - } - $this->table->clearMarks(); - $this->reload(); + $this->updateChrome(); + }); - if ($errors === []) { - $this->flash(sprintf('%s %s done', $action->label(), $this->describeTargets($targets)), Theme::SUCCESS); + 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->flash( - sprintf('%s failed: %s', $action->label(), $errors[0]) . (count($errors) > 1 ? sprintf(' (+%d more)', count($errors) - 1) : ''), - Theme::DANGER, + $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 { diff --git a/src/Console/Tui/Widget/FooterWidget.php b/src/Console/Tui/Widget/FooterWidget.php index 7fe7491e5..1f815b736 100644 --- a/src/Console/Tui/Widget/FooterWidget.php +++ b/src/Console/Tui/Widget/FooterWidget.php @@ -56,10 +56,10 @@ public function render(RenderContext $context): array Theme::SUCCESS => '✓', Theme::WARNING => '!', Theme::DANGER => '✗', - default => '•', + default => '', }; - $right = Theme::color($this->messageColor, $icon . ' ' . $this->message) . ' '; + $right = Theme::color($this->messageColor, ($icon === '' ? '' : $icon . ' ') . $this->message) . ' '; } if ($this->prompt !== null) { 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(); + } +} From c780619a3d0b0f4f589fd5d2b6f830529e901dfc Mon Sep 17 00:00:00 2001 From: David Badura Date: Wed, 7 Oct 2026 10:27:42 +0200 Subject: [PATCH 3/4] Add --stop-when-finished option to the run command The run worker stops as soon as all messages are processed. The dashboard uses it, so a run started there catches up completely instead of processing only one batch. --- docs/cli.md | 3 ++- .../Command/SubscriptionRunCommand.php | 16 ++++++++++++- .../Tui/ActionRunner/ProcessActionRunner.php | 2 +- .../Command/SubscriptionRunCommandTest.php | 23 +++++++++++++++++++ 4 files changed, 41 insertions(+), 3 deletions(-) diff --git a/docs/cli.md b/docs/cli.md index 229a7898d..1aea7cc0a 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -180,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 @@ -383,7 +384,7 @@ This requires the subscription commands to be registered in the same console app Otherwise, or with the `--in-process` option, the actions run in the dashboard process and block it until they are done. :::note -A run processes up to `--message-limit` messages once, it does not replace the [run worker](#run). +A run started from the dashboard stops as soon as all events are processed, it does not replace the [run worker](#run). ::: ### Keyboard shortcuts 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/ProcessActionRunner.php b/src/Console/Tui/ActionRunner/ProcessActionRunner.php index d4fb6685b..e99634230 100644 --- a/src/Console/Tui/ActionRunner/ProcessActionRunner.php +++ b/src/Console/Tui/ActionRunner/ProcessActionRunner.php @@ -84,7 +84,7 @@ private function arguments(SubscriptionAction $action): array return match ($action) { SubscriptionAction::Setup => [['event-sourcing:subscription:setup']], SubscriptionAction::Boot => [['event-sourcing:subscription:boot', ...$limit]], - SubscriptionAction::Run => [['event-sourcing:subscription:run', '--run-limit=1', '--sleep=0', ...$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']], 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); From 90a760ca55f24c6deab2153f924c6a491f2f02b3 Mon Sep 17 00:00:00 2001 From: David Badura Date: Wed, 7 Oct 2026 13:16:40 +0200 Subject: [PATCH 4/4] Show messages per second in the subscription dashboard The rate is calculated from the positions of the last reloads, averaged over ten seconds. It is shown as its own column, in the describe view together with an estimate of the remaining time, and for the head in the header. --- docs/cli.md | 2 +- src/Console/Tui/SubscriptionDashboard.php | 14 +++- src/Console/Tui/SubscriptionFormatter.php | 23 ++++- src/Console/Tui/Theme.php | 6 ++ src/Console/Tui/Throughput.php | 83 +++++++++++++++++++ .../Tui/View/SubscriptionTableView.php | 27 +++++- tests/Unit/Console/Tui/ThroughputTest.php | 59 +++++++++++++ 7 files changed, 206 insertions(+), 8 deletions(-) create mode 100644 src/Console/Tui/Throughput.php create mode 100644 tests/Unit/Console/Tui/ThroughputTest.php diff --git a/docs/cli.md b/docs/cli.md index 1aea7cc0a..f73ab5e51 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -335,7 +335,7 @@ You can find out more about the [subscription lifecycle](subscription.md#subscri ## 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 and lag, +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` diff --git a/src/Console/Tui/SubscriptionDashboard.php b/src/Console/Tui/SubscriptionDashboard.php index 28ecb3750..1272e1311 100644 --- a/src/Console/Tui/SubscriptionDashboard.php +++ b/src/Console/Tui/SubscriptionDashboard.php @@ -77,6 +77,7 @@ final class SubscriptionDashboard private array $pendingTargets = []; private int|null $head = null; + private readonly Throughput $headThroughput; private DateTimeImmutable|null $lastReload = null; private int $flashUntil = 0; @@ -97,6 +98,7 @@ public function __construct( $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()); @@ -146,6 +148,7 @@ public function reload(): void } $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()) { @@ -475,7 +478,12 @@ private function updateDescribe(): void $this->describe->update( sprintf('Describe(%s)', $subscription->id()), - SubscriptionFormatter::describe($subscription, $this->head, $this->lastReload ?? $this->now()), + SubscriptionFormatter::describe( + $subscription, + $this->head, + $this->lastReload ?? $this->now(), + $this->table->rate($subscription->id()), + ), ); } @@ -504,7 +512,9 @@ private function updateChrome(): void $meta[] = Theme::color(Theme::HIGHLIGHT, (string)$this->scope()); } - $meta[] = Theme::color(Theme::MUTED, 'head ') . Theme::color(Theme::TEXT, $this->head === null ? '–' : Theme::number($this->head)); + $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') ?? '–'); diff --git a/src/Console/Tui/SubscriptionFormatter.php b/src/Console/Tui/SubscriptionFormatter.php index b0dc80865..37f69c353 100644 --- a/src/Console/Tui/SubscriptionFormatter.php +++ b/src/Console/Tui/SubscriptionFormatter.php @@ -8,6 +8,7 @@ use Patchlevel\EventSourcing\Subscription\Subscription; use Patchlevel\EventSourcing\Subscription\SubscriptionError; +use function ceil; use function count; use function implode; use function intdiv; @@ -44,8 +45,11 @@ public static function age(DateTimeImmutable|null $time, DateTimeImmutable $now) return '-'; } - $seconds = max(0, $now->getTimestamp() - $time->getTimestamp()); + 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', @@ -60,7 +64,7 @@ public static function runMode(Subscription $subscription): string } /** @return list */ - public static function describe(Subscription $subscription, int|null $head, DateTimeImmutable $now): array + public static function describe(Subscription $subscription, int|null $head, DateTimeImmutable $now, float|null $rate = null): array { $error = $subscription->subscriptionError(); $status = Theme::status($subscription->status()); @@ -79,6 +83,7 @@ public static function describe(Subscription $subscription, int|null $head, Date 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())), @@ -130,6 +135,20 @@ private static function progress(Subscription $subscription, int|null $head): st . ($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 { diff --git a/src/Console/Tui/Theme.php b/src/Console/Tui/Theme.php index 77fab393c..a25527f2e 100644 --- a/src/Console/Tui/Theme.php +++ b/src/Console/Tui/Theme.php @@ -94,6 +94,12 @@ 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)); 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 index 2b07340e4..d3763b5f9 100644 --- a/src/Console/Tui/View/SubscriptionTableView.php +++ b/src/Console/Tui/View/SubscriptionTableView.php @@ -7,6 +7,7 @@ use DateTimeImmutable; use Patchlevel\EventSourcing\Console\Tui\SubscriptionFormatter; use Patchlevel\EventSourcing\Console\Tui\Theme; +use Patchlevel\EventSourcing\Console\Tui\Throughput; use Patchlevel\EventSourcing\Subscription\Subscription; use Symfony\Component\Tui\Ansi\AnsiUtils; @@ -34,9 +35,9 @@ */ final class SubscriptionTableView implements View { - private const COLUMNS = ['ID', 'GROUP', 'MODE', 'PROGRESS', 'POSITION', 'LAG', 'STATUS', 'RETRY', 'AGE', 'ERROR']; - private const RIGHT_ALIGNED = ['POSITION', 'LAG', 'RETRY', 'AGE']; - private const OPTIONAL = ['RETRY', 'MODE', 'GROUP', 'PROGRESS']; + private const COLUMNS = ['ID', 'GROUP', 'MODE', 'PROGRESS', 'POSITION', 'LAG', 'MSG/S', 'STATUS', 'RETRY', 'AGE', 'ERROR']; + private const RIGHT_ALIGNED = ['POSITION', 'LAG', 'MSG/S', 'RETRY', 'AGE']; + private const OPTIONAL = ['RETRY', 'MODE', 'GROUP', 'MSG/S', 'PROGRESS']; private const PROGRESS_BAR_WIDTH = 10; private const MIN_ERROR_WIDTH = 16; private const GAP = ' '; @@ -49,6 +50,7 @@ final class SubscriptionTableView implements View private int|null $head = null; private DateTimeImmutable $now; + private readonly Throughput $throughput; private string $filter = ''; private string|null $selectedId = null; private int $offset = 0; @@ -60,6 +62,7 @@ final class SubscriptionTableView implements View public function __construct() { $this->now = new DateTimeImmutable(); + $this->throughput = new Throughput(); } /** @param list $subscriptions */ @@ -73,6 +76,12 @@ public function update(array $subscriptions, int|null $head, DateTimeImmutable $ $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; @@ -101,6 +110,12 @@ 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) { @@ -305,6 +320,7 @@ 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()), @@ -320,6 +336,11 @@ private function cells(Subscription $subscription): array $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') 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')); + } +}