allow to refresh subscriptions

Author: DavidBadura

src/Console/Command/SubscriptionRefreshCommand.php

@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Console\Command;
+
+use LogicException;
+use Patchlevel\EventSourcing\Subscription\Engine\SubscriptionRefreshable;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+#[AsCommand(
+    'event-sourcing:subscription:refresh',
+    'Refresh subscriptions (run-mode, group)',
+)]
+final class SubscriptionRefreshCommand extends SubscriptionCommand
+{
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        if (!$this->engine instanceof SubscriptionRefreshable) {
+            throw new LogicException(sprintf(
+                '"%s" does not implement "%s" and can therefore not refresh subscriptions.',
+                $this->engine::class,
+                SubscriptionRefreshable::class,
+            ));
+        }
+
+        $criteria = $this->subscriptionEngineCriteria($input);
+        $this->engine->refreshSubscriptions($criteria);
+
+        return 0;
+    }
+}

src/Subscription/Engine/CatchUpSubscriptionEngine.php

@@ -4,13 +4,14 @@
 
 namespace Patchlevel\EventSourcing\Subscription\Engine;
 
+use LogicException;
 use Patchlevel\EventSourcing\Subscription\Subscription;
 
 use function array_merge;
 
 use const PHP_INT_MAX;
 
-final class CatchUpSubscriptionEngine implements SubscriptionEngine
+final class CatchUpSubscriptionEngine implements SubscriptionEngine, SubscriptionRefreshable
 {
     public function __construct(
         private readonly SubscriptionEngine $parent,
@@ -86,6 +87,19 @@ public function subscriptions(SubscriptionEngineCriteria|null $criteria = null):
         return $this->parent->subscriptions($criteria);
     }
 
+    public function refreshSubscriptions(SubscriptionEngineCriteria|null $criteria = null): Result
+    {
+        if (!$this->parent instanceof SubscriptionRefreshable) {
+            throw new LogicException(sprintf(
+                '"%s" does not implement "%s" and can therefore not refresh subscriptions.',
+                $this->parent::class,
+                SubscriptionRefreshable::class,
+            ));
+        }
+
+        return $this->parent->refreshSubscriptions($criteria);
+    }
+
     private function mergeResult(ProcessedResult ...$results): ProcessedResult
     {
         $processedMessages = 0;

src/Subscription/Engine/DefaultSubscriptionEngine.php

@@ -27,7 +27,7 @@
 use function count;
 use function sprintf;
 
-final class DefaultSubscriptionEngine implements SubscriptionEngine
+final class DefaultSubscriptionEngine implements SubscriptionEngine, SubscriptionRefreshable
 {
     private SubscriptionManager $subscriptionManager;
 
@@ -800,6 +800,68 @@ public function subscriptions(SubscriptionEngineCriteria|null $criteria = null):
         );
     }
 
+    public function refreshSubscriptions(SubscriptionEngineCriteria|null $criteria = null): Result
+    {
+        $criteria ??= new SubscriptionEngineCriteria();
+
+        $this->discoverNewSubscriptions();
+
+        $subscriptions = $this->subscriptionManager->find(new SubscriptionCriteria(
+            ids: $criteria->ids,
+            groups: $criteria->groups,
+        ));
+
+        foreach ($subscriptions as $subscription) {
+            $subscriber = $this->subscriber($subscription->id());
+
+            if (!$subscriber) {
+                continue;
+            }
+
+            $changed = false;
+
+            if ($subscription->runMode() !== $subscriber->runMode()) {
+                $changed = true;
+                $oldRunMode = $subscription->runMode();
+                $subscription->changeRunMode($subscriber->runMode());
+
+                $this->logger?->info(
+                    sprintf(
+                        'Subscription Engine: Subscription "%s" run mode changed from "%s" to "%s".',
+                        $subscription->id(),
+                        $oldRunMode->value,
+                        $subscription->runMode()->value,
+                    ),
+                );
+            }
+
+            if ($subscription->group() !== $subscriber->group()) {
+                $changed = true;
+                $oldGroup = $subscription->group();
+                $subscription->changeGroup($subscriber->group());
+
+                $this->logger?->info(
+                    sprintf(
+                        'Subscription Engine: Subscription "%s" group changed from "%s" to "%s".',
+                        $subscription->id(),
+                        $oldGroup,
+                        $subscription->group(),
+                    ),
+                );
+            }
+
+            if (!$changed) {
+                continue;
+            }
+
+            $this->subscriptionManager->update($subscription);
+        }
+
+        $this->subscriptionManager->flush();
+
+        return new Result();
+    }
+
     private function handleMessage(int $index, Message $message, Subscription $subscription): Error|null
     {
         $subscriber = $this->subscriber($subscription->id());

src/Subscription/Engine/SubscriptionRefreshable.php

@@ -0,0 +1,8 @@
+<?php
+
+namespace Patchlevel\EventSourcing\Subscription\Engine;
+
+interface SubscriptionRefreshable
+{
+    public function refreshSubscriptions(SubscriptionEngineCriteria|null $criteria = null): Result;
+}

src/Subscription/Engine/ThrowOnErrorSubscriptionEngine.php

@@ -4,9 +4,10 @@
 
 namespace Patchlevel\EventSourcing\Subscription\Engine;
 
+use LogicException;
 use Patchlevel\EventSourcing\Subscription\Subscription;
 
-final class ThrowOnErrorSubscriptionEngine implements SubscriptionEngine
+final class ThrowOnErrorSubscriptionEngine implements SubscriptionEngine, SubscriptionRefreshable
 {
     public function __construct(
         private readonly SubscriptionEngine $parent,
@@ -54,6 +55,19 @@ public function subscriptions(SubscriptionEngineCriteria|null $criteria = null):
         return $this->parent->subscriptions($criteria);
     }
 
+    public function refreshSubscriptions(SubscriptionEngineCriteria|null $criteria = null): Result
+    {
+        if (!$this->parent instanceof SubscriptionRefreshable) {
+            throw new LogicException(sprintf(
+                '"%s" does not implement "%s" and can therefore not refresh subscriptions.',
+                $this->parent::class,
+                SubscriptionRefreshable::class,
+            ));
+        }
+
+        return $this->throwOnError($this->parent->refreshSubscriptions($criteria));
+    }
+
     /**
      * @param T $result
      *

src/Subscription/Subscription.php

@@ -13,8 +13,8 @@ final class Subscription
 
     public function __construct(
         private readonly string $id,
-        private readonly string $group = self::DEFAULT_GROUP,
-        private readonly RunMode $runMode = RunMode::FromBeginning,
+        private string $group = self::DEFAULT_GROUP,
+        private RunMode $runMode = RunMode::FromBeginning,
         private Status $status = Status::New,
         private int $position = 0,
         private SubscriptionError|null $error = null,
@@ -33,11 +33,21 @@ public function group(): string
         return $this->group;
     }
 
+    public function changeGroup(string $group): void
+    {
+        $this->group = $group;
+    }
+
     public function runMode(): RunMode
     {
         return $this->runMode;
     }
 
+    public function changeRunMode(RunMode $runMode): void
+    {
+        $this->runMode = $runMode;
+    }
+
     public function status(): Status
     {
         return $this->status;