Add tag filtering support for event replay

src/Attributes/Hooks/Tag.php

@@ -0,0 +1,17 @@
+<?php
+
+namespace Thunk\Verbs\Attributes\Hooks;
+
+use Attribute;
+
+#[Attribute(Attribute::TARGET_METHOD)]
+class Tag
+{
+    /** @var string[] */
+    public array $tags;
+
+    public function __construct(string|array $tag)
+    {
+        $this->tags = is_array($tag) ? $tag : [$tag];
+    }
+}

src/Commands/ReplayCommand.php

@@ -18,7 +18,7 @@
 
 class ReplayCommand extends Command
 {
-    protected $signature = 'verbs:replay {--force}';
+    protected $signature = 'verbs:replay {--force} {--tag=*}';
 
     protected $description = 'Replay all Verbs events.';
 
@@ -28,14 +28,26 @@ public function handle(BrokersEvents $broker): int
             return 1;
         }
 
+        /** @var string[] $tags */
+        $tags = $this->option('tag');
+        $tags = array_map('strtolower', $tags);
+
         // Prepare for a long-running, database-heavy run
         ini_set('memory_limit', '-1');
         EventFacade::forget(QueryExecuted::class);
         DB::disableQueryLog();
 
+        $count = VerbEvent::count();
+
+        if ($count === 0) {
+            $this->info('No events to replay.');
+
+            return 0;
+        }
+
         $started_at = time();
 
-        $progress = progress('Replaying…', VerbEvent::count());
+        $progress = progress('Replaying…', $count);
         $progress->start();
 
         $broker->replay(
@@ -46,6 +58,7 @@ class_basename($event),
                 $event->id,
             )),
             afterEach: fn () => $progress->advance(),
+            tags: $tags,
         );
 
         $progress->finish();

src/Contracts/BrokersEvents.php

@@ -14,5 +14,5 @@ public function isAuthorized(Event $event): bool;
 
     public function isValid(Event $event): bool;
 
-    public function replay(?callable $beforeEach = null, ?callable $afterEach = null);
+    public function replay(?callable $beforeEach = null, ?callable $afterEach = null, ?array $tags = null);
 }

src/Lifecycle/Broker.php

@@ -77,9 +77,10 @@ public function commit(): bool
         return $this->commit();
     }
 
-    public function replay(?callable $beforeEach = null, ?callable $afterEach = null): void
+    public function replay(?callable $beforeEach = null, ?callable $afterEach = null, ?array $tags = null): void
     {
         $this->is_replaying = true;
+        $this->replay_include_tags = $tags;
 
         try {
             $this->states->reset(include_storage: true);
@@ -110,6 +111,7 @@ public function replay(?callable $beforeEach = null, ?callable $afterEach = null
             $this->states->writeSnapshots();
             $this->states->prune();
             $this->states->setReplaying(false);
+            $this->replay_include_tags = null;
             $this->is_replaying = false;
         }
     }

src/Lifecycle/BrokerConvenienceMethods.php

@@ -18,6 +18,8 @@ trait BrokerConvenienceMethods
 {
     public bool $is_replaying = false;
 
+    public ?array $replay_include_tags = null;
+
     /**
      * @deprecated
      * @see IdManager

src/Lifecycle/Dispatcher.php

@@ -118,23 +118,15 @@ protected function getHandleHooks(Event $event): Collection
     {
         $hooks = $this->hooksFor($event, Phase::Handle);
 
-        if (method_exists($event, 'handle')) {
-            $hooks->prepend(Hook::fromClassMethod($event, 'handle')->forcePhases(Phase::Handle, Phase::Replay));
-        }
-
-        return $hooks;
+        return $hooks->merge($this->hooksWithPrefix($event, Phase::Handle, 'handle'));
     }
 
     /** @return Collection<int, Hook> */
     protected function getReplayHooks(Event $event): Collection
     {
         $hooks = $this->hooksFor($event, Phase::Replay);
 
-        if (method_exists($event, 'handle')) {
-            $hooks->prepend(Hook::fromClassMethod($event, 'handle')->forcePhases(Phase::Handle, Phase::Replay));
-        }
-
-        return $hooks;
+        return $hooks->merge($this->hooksWithPrefix($event, Phase::Replay, 'handle'));
     }
 
     /** @return Collection<int, Hook> */

src/Lifecycle/Hook.php

@@ -19,11 +19,20 @@ public static function fromClassMethod(object $target, ReflectionMethod|string $
         if (is_string($method)) {
             $method = new ReflectionMethod($target, $method);
         }
+        $tagAttributes = $method->getAttributes(\Thunk\Verbs\Attributes\Hooks\Tag::class);
+
+        $tags = collect($tagAttributes)
+            ->map(fn ($attr) => $attr->newInstance())
+            ->map(fn ($tag) => $tag->tags)
+            ->flatten()
+            ->map(fn ($tag) => strtolower($tag))
+            ->all();
 
         $hook = new static(
             callback: Closure::fromCallable([$target, $method->getName()]),
             targets: Reflector::getParameterTypes($method),
             name: $method->getName(),
+            tags: $tags,
         );
 
         return Reflector::applyHookAttributes($method, $hook);
@@ -44,12 +53,15 @@ public function __construct(
         public array $targets = [],
         public SplObjectStorage $phases = new SplObjectStorage,
         public ?string $name = null,
+        public ?array $tags = null,
     ) {}
 
     public function forcePhases(Phase ...$phases): static
     {
         foreach ($phases as $phase) {
-            $this->phases[$phase] = true;
+            if (! isset($this->phases[$phase])) {
+                $this->phases[$phase] = true;
+            }
         }
 
         return $this;
@@ -112,6 +124,12 @@ public function handle(Container $container, Event $event): mixed
 
     public function replay(Container $container, Event $event): void
     {
+        if ($filteringTags = app(Broker::class)->replay_include_tags) {
+            if (empty(array_intersect($filteringTags, $this->tags))) {
+                return;
+            }
+        }
+
         if ($this->runsInPhase(Phase::Replay)) {
             app(Wormhole::class)->warp($event, fn () => $this->execute($container, $event));
         }

src/Testing/BrokerFake.php

@@ -72,9 +72,9 @@ public function commit(): bool
         return $this->broker->commit();
     }
 
-    public function replay(?callable $beforeEach = null, ?callable $afterEach = null)
+    public function replay(?callable $beforeEach = null, ?callable $afterEach = null, ?array $tags = null)
     {
-        $this->broker->replay($beforeEach, $afterEach);
+        $this->broker->replay($beforeEach, $afterEach, $tags);
     }
 
     public function commitImmediately(bool $commit_immediately = true): void

tests/Feature/ReplayCommandTest.php

@@ -3,6 +3,8 @@
 use Carbon\CarbonImmutable;
 use Illuminate\Support\Carbon;
 use Thunk\Verbs\Attributes\Autodiscovery\StateId;
+use Thunk\Verbs\Attributes\Hooks\Once;
+use Thunk\Verbs\Attributes\Hooks\Tag;
 use Thunk\Verbs\Commands\ReplayCommand;
 use Thunk\Verbs\Event;
 use Thunk\Verbs\Facades\Id;
@@ -45,7 +47,7 @@
         ->toBe(4)
         ->and($GLOBALS['replay_test_counts'][$state2_id])
         ->toBe(4)
-        ->and($GLOBALS['handle_count'])->toBe(10);
+        ->and($GLOBALS['handle_count'])->toBe(10 * 2);
 
     // Reset 'projected' state and change data that only is touched when not replaying
     $GLOBALS['replay_test_counts'] = [];
@@ -65,6 +67,13 @@
         ->and($GLOBALS['handle_count'])->toBe(1337);
 });
 
+it('can replay with no events', function () {
+    config(['app.env' => 'testing']);
+    $this->artisan(ReplayCommand::class);
+
+    expect(Thunk\Verbs\Models\VerbEvent::count())->toBe(0);
+});
+
 it('uses the original event times when replaying', function () {
     \Illuminate\Support\Facades\Date::setTestNow('2024-04-01 12:00:00');
     $state_id = Id::make();
@@ -136,6 +145,81 @@
     expect($snapshot2->created_at)->toEqual(CarbonImmutable::parse('2024-05-15 18:00:00'));
 });
 
+it('can filter replayed events by tags', function () {
+    $GLOBALS['email_sent'] = [];
+    $GLOBALS['notification_sent'] = [];
+    $GLOBALS['billing_processed'] = [];
+
+    // Fire events with different tagged methods
+    TaggedReplayEvent::fire(state_id: Id::make());
+    TaggedReplayEvent::fire(state_id: Id::make());
+    TaggedReplayEvent::fire(state_id: Id::make());
+
+    Verbs::commit();
+
+    // Verify initial state
+    expect($GLOBALS['email_sent'])->toHaveCount(3)
+        ->and($GLOBALS['notification_sent'])->toHaveCount(3)
+        ->and($GLOBALS['billing_processed'])->toHaveCount(3);
+
+    // Reset counters
+    $GLOBALS['email_sent'] = [];
+    $GLOBALS['notification_sent'] = [];
+    $GLOBALS['billing_processed'] = [];
+
+    // Test single tag filter
+    config(['app.env' => 'testing']);
+    $this->artisan(ReplayCommand::class, ['--tag' => ['email']]);
+
+    expect($GLOBALS['email_sent'])->toHaveCount(3)
+        ->and($GLOBALS['notification_sent'])->toHaveCount(0)
+        ->and($GLOBALS['billing_processed'])->toHaveCount(0);
+
+    // Reset counters
+    $GLOBALS['email_sent'] = [];
+    $GLOBALS['notification_sent'] = [];
+    $GLOBALS['billing_processed'] = [];
+
+    // Test multiple tags
+    $this->artisan(ReplayCommand::class, ['--tag' => ['email', 'billing']]);
+
+    expect($GLOBALS['email_sent'])->toHaveCount(3)
+        ->and($GLOBALS['notification_sent'])->toHaveCount(0)
+        ->and($GLOBALS['billing_processed'])->toHaveCount(3);
+
+    // Reset counters
+    $GLOBALS['email_sent'] = [];
+    $GLOBALS['notification_sent'] = [];
+    $GLOBALS['billing_processed'] = [];
+
+    // Test with important tag
+    $this->artisan(ReplayCommand::class, ['--tag' => ['important']]);
+
+    expect($GLOBALS['email_sent'])->toHaveCount(0)
+        ->and($GLOBALS['notification_sent'])->toHaveCount(0)
+        ->and($GLOBALS['billing_processed'])->toHaveCount(3);
+});
+
+it('handles case sensitivity in tags correctly', function () {
+    $GLOBALS['email_sent'] = [];
+    $GLOBALS['notification_sent'] = [];
+    $GLOBALS['billing_processed'] = [];
+
+    TaggedReplayEvent::fire(state_id: Id::make());
+    Verbs::commit();
+
+    $GLOBALS['email_sent'] = [];
+    $GLOBALS['notification_sent'] = [];
+    $GLOBALS['billing_processed'] = [];
+
+    config(['app.env' => 'testing']);
+    $this->artisan(ReplayCommand::class, ['--tag' => ['EMAIL']]);
+
+    expect($GLOBALS['email_sent'])->toHaveCount(1)
+        ->and($GLOBALS['notification_sent'])->toHaveCount(0)
+        ->and($GLOBALS['billing_processed'])->toHaveCount(0);
+});
+
 class ReplayCommandTestEvent extends Event
 {
     public function __construct(
@@ -158,6 +242,12 @@ public function handle()
 
         Verbs::unlessReplaying(fn () => $GLOBALS['handle_count']++);
     }
+
+    #[Once]
+    public function handleTwo()
+    {
+        $GLOBALS['handle_count']++;
+    }
 }
 
 class ReplayCommandTestState extends State
@@ -186,3 +276,33 @@ class ReplayCommandTestWormholeState extends State
 {
     public CarbonImmutable $time;
 }
+
+class TaggedReplayEvent extends Event
+{
+    public function __construct(
+        #[StateId(TaggedReplayState::class)] public ?int $state_id = null,
+    ) {}
+
+    #[Tag('email')]
+    public function handleSendEmail()
+    {
+        $GLOBALS['email_sent'][] = $this->id;
+    }
+
+    #[Tag('notification')]
+    public function handleSendNotification()
+    {
+        $GLOBALS['notification_sent'][] = $this->id;
+    }
+
+    #[Tag(['billing', 'important'])]
+    public function handleProcessBilling()
+    {
+        $GLOBALS['billing_processed'][] = $this->id;
+    }
+}
+
+class TaggedReplayState extends State
+{
+    public int $count = 0;
+}