POC: add subscriber delay

src/Attribute/Delay.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Attribute;
+
+use Attribute;
+use DateInterval;
+use InvalidArgumentException;
+
+#[Attribute(Attribute::TARGET_CLASS)]
+final class Delay
+{
+    public readonly DateInterval $delay;
+
+    public function __construct(
+        string $dateString,
+    ) {
+        $interval = DateInterval::createFromDateString($dateString);
+
+        if ($interval === false) {
+            throw new InvalidArgumentException('Invalid date string');
+        }
+
+        $this->delay = $interval;
+    }
+}

src/Metadata/Subscriber/AttributeSubscriberMetadataFactory.php

@@ -4,6 +4,8 @@
 
 namespace Patchlevel\EventSourcing\Metadata\Subscriber;
 
+use DateInterval;
+use Patchlevel\EventSourcing\Attribute\Delay;
 use Patchlevel\EventSourcing\Attribute\Setup;
 use Patchlevel\EventSourcing\Attribute\Subscribe;
 use Patchlevel\EventSourcing\Attribute\Subscriber;
@@ -87,6 +89,7 @@ public function metadata(string $subscriber): SubscriberMetadata
             $subscribeMethods,
             $setupMethod,
             $teardownMethod,
+            $this->delay($reflector),
         );
 
         $this->subscriberMetadata[$subscriber] = $metadata;
@@ -128,4 +131,17 @@ private function subscribeMethod(ReflectionMethod $method): SubscribeMethodMetad
             $arguments,
         );
     }
+
+    private function delay(ReflectionClass $reflector): DateInterval|null
+    {
+        $attributes = $reflector->getAttributes(Delay::class);
+
+        if ($attributes === []) {
+            return null;
+        }
+
+        $instance = $attributes[0]->newInstance();
+
+        return $instance->delay;
+    }
 }

src/Metadata/Subscriber/SubscriberMetadata.php

@@ -4,6 +4,7 @@
 
 namespace Patchlevel\EventSourcing\Metadata\Subscriber;
 
+use DateInterval;
 use Patchlevel\EventSourcing\Subscription\RunMode;
 use Patchlevel\EventSourcing\Subscription\Subscription;
 
@@ -17,6 +18,7 @@ public function __construct(
         public readonly array $subscribeMethods = [],
         public readonly string|null $setupMethod = null,
         public readonly string|null $teardownMethod = null,
+        public readonly DateInterval|null $delay = null,
     ) {
     }
 }

src/Subscription/Engine/DefaultSubscriptionEngine.php

@@ -4,19 +4,26 @@
 
 namespace Patchlevel\EventSourcing\Subscription\Engine;
 
+use DateTimeImmutable;
+use Patchlevel\EventSourcing\Aggregate\AggregateHeader;
+use Patchlevel\EventSourcing\Clock\SystemClock;
+use Patchlevel\EventSourcing\Message\HeaderNotFound;
 use Patchlevel\EventSourcing\Message\Message;
 use Patchlevel\EventSourcing\Store\Store;
+use Patchlevel\EventSourcing\Store\StreamHeader;
 use Patchlevel\EventSourcing\Subscription\RetryStrategy\ClockBasedRetryStrategy;
 use Patchlevel\EventSourcing\Subscription\RetryStrategy\RetryStrategy;
 use Patchlevel\EventSourcing\Subscription\RunMode;
 use Patchlevel\EventSourcing\Subscription\Status;
 use Patchlevel\EventSourcing\Subscription\Store\SubscriptionCriteria;
 use Patchlevel\EventSourcing\Subscription\Store\SubscriptionStore;
 use Patchlevel\EventSourcing\Subscription\Subscriber\BatchableSubscriber;
+use Patchlevel\EventSourcing\Subscription\Subscriber\MetadataSubscriberAccessor;
 use Patchlevel\EventSourcing\Subscription\Subscriber\RealSubscriberAccessor;
 use Patchlevel\EventSourcing\Subscription\Subscriber\SubscriberAccessor;
 use Patchlevel\EventSourcing\Subscription\Subscriber\SubscriberAccessorRepository;
 use Patchlevel\EventSourcing\Subscription\Subscription;
+use Psr\Clock\ClockInterface;
 use Psr\Log\LoggerInterface;
 use Throwable;
 
@@ -41,6 +48,7 @@ public function __construct(
         private readonly SubscriberAccessorRepository $subscriberRepository,
         private readonly RetryStrategy $retryStrategy = new ClockBasedRetryStrategy(),
         private readonly LoggerInterface|null $logger = null,
+        private readonly ClockInterface $clock = new SystemClock(),
     ) {
         if ($messageStore instanceof MessageLoader) {
             $this->messageLoader = $messageStore;
@@ -224,6 +232,32 @@ function (SubscriptionCollection $subscriptions) use ($limit): ProcessedResult {
                                     continue;
                                 }
 
+                                if ($this->shouldDelay($subscription, $message)) {
+                                    $subscriptions->remove($subscription);
+
+                                    if ($subscription->runMode() === RunMode::Once) {
+                                        $subscription->finished();
+                                        $this->subscriptionManager->update($subscription);
+
+                                        $this->logger?->info(sprintf(
+                                            'Subscription Engine: Subscription "%s" reached delayed message, set to finished.',
+                                            $subscription->id(),
+                                        ));
+
+                                        continue;
+                                    }
+
+                                    $subscription->active();
+                                    $this->subscriptionManager->update($subscription);
+
+                                    $this->logger?->info(sprintf(
+                                        'Subscription Engine: Subscription "%s" reached delayed message, set to active.',
+                                        $subscription->id(),
+                                    ));
+
+                                    continue;
+                                }
+
                                 $error = $this->handleMessage($index, $message, $subscription);
 
                                 if (!$error) {
@@ -395,6 +429,19 @@ function (SubscriptionCollection $subscriptions) use ($limit): ProcessedResult {
                                     continue;
                                 }
 
+                                if ($this->shouldDelay($subscription, $message)) {
+                                    $subscriptions->remove($subscription);
+
+                                    $this->logger?->debug(
+                                        sprintf(
+                                            'Subscription Engine: Subscription "%s" is delayed, skip processing.',
+                                            $subscription->id(),
+                                        ),
+                                    );
+
+                                    continue;
+                                }
+
                                 $error = $this->handleMessage($index, $message, $subscription);
 
                                 if (!$error) {
@@ -1109,4 +1156,38 @@ private function shouldCommitBatch(Subscription $subscription): bool
 
         return $this->batching[$subscription->id()]->forceCommit();
     }
+
+    private function shouldDelay(Subscription $subscription, Message $message): bool
+    {
+        $subscriber = $this->subscriber($subscription->id());
+
+        if (!$subscriber instanceof MetadataSubscriberAccessor) {
+            return false;
+        }
+
+        $delay = $subscriber->metadata()->delay;
+
+        if ($delay === null) {
+            return false;
+        }
+
+        $recordedOn = $this->recordedOn($message);
+
+        if ($recordedOn === null) {
+            return false;
+        }
+
+        $threshold = $this->clock->now()->sub($delay);
+
+        return $recordedOn <= $threshold;
+    }
+
+    private function recordedOn(Message $message): DateTimeImmutable|null
+    {
+        try {
+            return $message->header(AggregateHeader::class)->recordedOn;
+        } catch (HeaderNotFound) {
+            return $message->header(StreamHeader::class)->recordedOn;
+        }
+    }
 }