Send events from the domain itself

Currently, the application layer is responsible for sending events that
are actually more related to the domain. For example, the `UpsertArticleHandler`
publishes an event when an article is created.

In this commit we move the publishing of domain events to the entity
itself. By adding a doctrine listener we make sure the events are only
emitted when the entity is committed to the database.

src/Common/Domain/EventPublishingAggregateRoot.php

@@ -0,0 +1,12 @@
+<?php
+
+namespace App\Common\Domain;
+
+interface EventPublishingAggregateRoot
+{
+    public function recordEvent(object $event): void;
+
+    public function shiftDomainEvent(): object;
+
+    public function hasDomainEvents(): bool;
+}

src/Common/Domain/EventPublishingAggregateRootTrait.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace App\Common\Domain;
+
+trait EventPublishingAggregateRootTrait
+{
+    /**
+     * @var list<object>
+     */
+    private array $domainEvents = [];
+
+    public function recordEvent(object $event): void
+    {
+        $this->domainEvents[] = $event;
+    }
+
+    public function shiftDomainEvent(): object
+    {
+        return array_shift($this->domainEvents) ?? throw new \LogicException('There are no domain events (left).');
+    }
+
+    public function hasDomainEvents(): bool
+    {
+        return $this->domainEvents !== [];
+    }
+}

src/Common/Infrastructure/Persistence/Doctrine/EmitDomainEventsOnFlushListener.php

@@ -0,0 +1,57 @@
+<?php
+
+namespace App\Common\Infrastructure\Persistence\Doctrine;
+
+use App\Common\Domain\EventPublishingAggregateRoot;
+use App\Common\Infrastructure\Messenger\EventBus\EventBus;
+use Doctrine\Bundle\DoctrineBundle\Attribute\AsEntityListener;
+use Doctrine\ORM\Event\OnFlushEventArgs;
+use Doctrine\ORM\Events;
+
+#[AsEntityListener(event: Events::onFlush, method: 'onFlush')]
+final readonly class EmitDomainEventsOnFlushListener
+{
+    public function __construct(private EventBus $eventBus)
+    {
+    }
+
+    public function onFlush(OnFlushEventArgs $event): void
+    {
+        $uow = $event->getObjectManager()->getUnitOfWork();
+
+        foreach ($uow->getScheduledEntityInsertions() as $entity) {
+            $this->emitRecordedEvents($entity);
+        }
+
+        foreach ($uow->getScheduledEntityUpdates() as $entity) {
+            $this->emitRecordedEvents($entity);
+        }
+
+        foreach ($uow->getScheduledEntityDeletions() as $entity) {
+            $this->emitRecordedEvents($entity);
+        }
+
+        foreach ($uow->getScheduledCollectionDeletions() as $collection) {
+            foreach ($collection as $entity) {
+                $this->emitRecordedEvents($entity);
+            }
+        }
+
+        foreach ($uow->getScheduledCollectionUpdates() as $collection) {
+            foreach ($collection as $entity) {
+                $this->emitRecordedEvents($entity);
+            }
+        }
+    }
+
+    public function emitRecordedEvents(object $entity): void
+    {
+        if ($entity instanceof EventPublishingAggregateRoot === false) {
+            return;
+        }
+
+        while ($entity->hasDomainEvents()) {
+            $this->eventBus->dispatch($entity->shiftDomainEvent());
+        }
+    }
+}

src/Feed/Application/Command/Article/Handler/UpsertArticleHandler.php

@@ -3,21 +3,15 @@
 namespace App\Feed\Application\Command\Article\Handler;
 
 use App\Common\Infrastructure\Messenger\CommandBus\AsCommandHandler;
-use App\Common\Infrastructure\Messenger\EventBus\EventBus;
 use App\Feed\Application\Command\Article\UpsertArticleCommand;
-use App\Feed\Application\Event\Article\ArticleAddedEvent;
-use App\Feed\Application\Event\Article\ArticleUpdatedEvent;
 use App\Feed\Domain\Article\Article;
 use App\Feed\Domain\Article\ArticleId;
 use App\Feed\Domain\Article\ArticleRepository;
 use App\Feed\Domain\Article\Url\Exception\MalformedUrlException;
 use App\Feed\Domain\Article\Url\Url;
 use App\Feed\Domain\Source\Exception\SourceNotFoundException;
-use App\Feed\Domain\Source\SourceId;
 use App\Feed\Domain\Source\SourceRepository;
-use Psr\Log\LoggerInterface;
 use Ramsey\Uuid\Uuid;
-use Symfony\Component\Messenger\Attribute\AsMessageHandler;
 
 /**
  * Updates the article if the url is found in the article repository, otherwise it creates a new article.
@@ -28,7 +22,6 @@
     public function __construct(
         private ArticleRepository $articleRepository,
         private SourceRepository $sourceRepository,
-        private EventBus $eventBus,
     ) {
     }
 
@@ -56,8 +49,6 @@ public function __invoke(UpsertArticleCommand $command): void
 
             $this->articleRepository->save($article);
 
-            $this->eventBus->dispatch(new ArticleAddedEvent($article->getId()));
-
             return;
         }
 
@@ -70,7 +61,5 @@ public function __invoke(UpsertArticleCommand $command): void
         );
 
         $this->articleRepository->save($existingArticle);
-
-        $this->eventBus->dispatch(new ArticleUpdatedEvent($existingArticle->getId()));
     }
 }

src/Feed/Application/Listener/Article/InvalidateArticleCacheListener.php

@@ -3,7 +3,7 @@
 namespace App\Feed\Application\Listener\Article;
 
 use App\Common\Infrastructure\Messenger\EventBus\AsEventSubscriber;
-use App\Feed\Application\Event\Article\ArticleUpdatedEvent;
+use App\Feed\Domain\Article\Event\Article\ArticleUpdatedEvent;
 use App\Feed\Infrastructure\Cache\FeedCacheKeys;
 use Symfony\Contracts\Cache\CacheInterface;
 

src/Feed/Application/Listener/Article/InvalidateTotalArticlesCountCacheListener.php

@@ -3,8 +3,7 @@
 namespace App\Feed\Application\Listener\Article;
 
 use App\Common\Infrastructure\Messenger\EventBus\AsEventSubscriber;
-use App\Feed\Application\Event\Article\ArticleAddedEvent;
-use App\Feed\Domain\Article\ArticleRepository;
+use App\Feed\Domain\Article\Event\Article\ArticleAddedEvent;
 use App\Feed\Infrastructure\Cache\FeedCacheKeys;
 use Symfony\Contracts\Cache\CacheInterface;
 

src/Feed/Domain/Article/Article.php

@@ -2,6 +2,10 @@
 
 namespace App\Feed\Domain\Article;
 
+use App\Common\Domain\EventPublishingAggregateRoot;
+use App\Common\Domain\EventPublishingAggregateRootTrait;
+use App\Feed\Domain\Article\Event\Article\ArticleAddedEvent;
+use App\Feed\Domain\Article\Event\Article\ArticleUpdatedEvent;
 use App\Feed\Domain\Article\Url\Url;
 use App\Feed\Domain\Source\Source;
 use DateTime;
@@ -10,8 +14,10 @@
 #[ORM\Entity]
 #[ORM\Table(name: 'articles')]
 #[ORM\Index(columns: ['url'], name: 'url_index')]
-class Article
+class Article implements EventPublishingAggregateRoot
 {
+    use EventPublishingAggregateRootTrait;
+
     #[ORM\Id, ORM\Column(type: 'string')]
     private string $id;
 
@@ -44,6 +50,8 @@ public function __construct(
         $this->url = (string) $url;
         $this->updated = $updated;
         $this->source = $source;
+
+        $this->recordEvent(new ArticleAddedEvent($id));
     }
 
     public function getId(): ArticleId
@@ -91,5 +99,7 @@ public function updateArticle(string $newTitle, string $newSummary, Url $newUrl,
         $this->url = $newUrl;
         $this->updated = $newUpdated;
         $this->source = $newSource;
+
+        $this->recordEvent(new ArticleUpdatedEvent($this->id));
     }
 }

src/Feed/Application/Event/Article/ArticleAddedEvent.php → src/Feed/Domain/Article/Event/Article/ArticleAddedEvent.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace App\Feed\Application\Event\Article;
+namespace App\Feed\Domain\Article\Event\Article;
 
 final readonly class ArticleAddedEvent
 {

src/Feed/Application/Event/Article/ArticleUpdatedEvent.php → src/Feed/Domain/Article/Event/Article/ArticleUpdatedEvent.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace App\Feed\Application\Event\Article;
+namespace App\Feed\Domain\Article\Event\Article;
 
 final readonly class ArticleUpdatedEvent
 {

tests/Unit/Feed/Application/Command/Article/Handler/UpsertArticleHandlerTest.php

@@ -2,23 +2,18 @@
 
 namespace Unit\Feed\Application\Command\Article\Handler;
 
-use App\Feed\Application\Command\Article\UpsertArticleCommand;
 use App\Feed\Application\Command\Article\Handler\UpsertArticleHandler;
-use App\Feed\Application\Event\Article\ArticleAddedEvent;
-use App\Feed\Application\Event\Article\ArticleUpdatedEvent;
+use App\Feed\Application\Command\Article\UpsertArticleCommand;
 use App\Feed\Domain\Article\ArticleId;
 use App\Feed\Domain\Article\Url\Exception\MalformedUrlException;
 use App\Feed\Domain\Article\Url\Exception\SchemeNotSupportedException;
 use App\Feed\Domain\Source\Exception\SourceNotFoundException;
-use App\Feed\Domain\Source\SourceId;
 use DateTime;
-use Dev\Common\Infrastructure\Messenger\EventBus\RecordingEventBus;
 use Dev\Feed\Factory\ArticleFactory;
 use Dev\Feed\Factory\SourceFactory;
 use Dev\Feed\Repository\InMemoryArticleRepository;
 use Dev\Feed\Repository\InMemorySourceRepository;
 use PHPUnit\Framework\TestCase;
-use Psr\Log\LogLevel;
 use Ramsey\Uuid\Uuid;
 
 final class UpsertArticleHandlerTest extends TestCase
@@ -27,18 +22,14 @@ final class UpsertArticleHandlerTest extends TestCase
     private InMemoryArticleRepository $articleRepository;
     private InMemorySourceRepository $sourceRepository;
 
-    private RecordingEventBus $eventBus;
-
     public function setUp(): void
     {
         $this->articleRepository = new InMemoryArticleRepository();
         $this->sourceRepository = new InMemorySourceRepository();
-        $this->eventBus = new RecordingEventBus();
 
         $this->handler = new UpsertArticleHandler(
             $this->articleRepository,
             $this->sourceRepository,
-            $this->eventBus,
         );
     }
 
@@ -78,10 +69,6 @@ public function it_should_create_an_article(): void
         self::assertSame($url, (string) $article->getUrl());
         self::assertSame($updated, $article->getUpdated());
         self::assertSame($source, $article->getSource());
-
-        $event = $this->eventBus->shiftEvent();
-        self::assertInstanceOf(ArticleAddedEvent::class, $event);
-        self::assertSame((string) $article->getId(), $event->articleId);
     }
 
     /**
@@ -120,10 +107,6 @@ public function it_should_update_the_article(): void
         self::assertEquals($existingArticle->getUrl(), $article->getUrl());
         self::assertSame($updated, $article->getUpdated());
         self::assertSame($source, $article->getSource());
-
-        $event = $this->eventBus->shiftEvent();
-        self::assertInstanceOf(ArticleUpdatedEvent::class, $event);
-        self::assertSame((string) $article->getId(), $event->articleId);
     }
 
     /**

tests/Unit/Feed/Application/Listener/Article/InvalidateArticleCacheListenerTest.php

@@ -2,13 +2,12 @@
 
 namespace Unit\Feed\Application\Listener\Article;
 
-use App\Feed\Application\Event\Article\ArticleUpdatedEvent;
 use App\Feed\Application\Listener\Article\InvalidateArticleCacheListener;
+use App\Feed\Domain\Article\Event\Article\ArticleUpdatedEvent;
 use App\Feed\Infrastructure\Cache\FeedCacheKeys;
 use Dev\Feed\Factory\ArticleFactory;
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\Cache\Adapter\ArrayAdapter;
-use Symfony\Contracts\Cache\CacheInterface;
 use Symfony\Contracts\Cache\ItemInterface;
 
 final class InvalidateArticleCacheListenerTest extends TestCase

tests/Unit/Feed/Domain/Article/ArticleTest.php

@@ -2,6 +2,8 @@
 
 namespace Unit\Feed\Domain\Article;
 
+use App\Feed\Domain\Article\Event\Article\ArticleAddedEvent;
+use App\Feed\Domain\Article\Event\Article\ArticleUpdatedEvent;
 use Dev\Feed\Factory\ArticleFactory;
 use PHPUnit\Framework\TestCase;
 
@@ -32,6 +34,9 @@ public function it_should_not_update_the_article_if_the_updates_are_older(): voi
         self::assertEquals('existing title', $article->getTitle());
         self::assertEquals('existing summary', $article->getSummary());
         self::assertEquals(new \DateTime('2022-03-03 00:00:00'), $article->getUpdated());
+
+        self::assertEquals(new ArticleAddedEvent($article->getId()), $article->shiftDomainEvent());
+        self::assertFalse($article->hasDomainEvents());
     }
 
     /**
@@ -59,6 +64,9 @@ public function it_should_not_update_the_article_if_the_updates_happened_at_the_
         self::assertEquals('existing title', $article->getTitle());
         self::assertEquals('existing summary', $article->getSummary());
         self::assertEquals(new \DateTime('2022-03-03 00:00:00'), $article->getUpdated());
+
+        self::assertEquals(new ArticleAddedEvent($article->getId()), $article->shiftDomainEvent());
+        self::assertFalse($article->hasDomainEvents());
     }
 
     /**
@@ -86,5 +94,9 @@ public function it_should_update_the_article_if_the_updates_are_newer(): void
         self::assertEquals('updated title', $article->getTitle());
         self::assertEquals('updated summary', $article->getSummary());
         self::assertEquals(new \DateTime('2024-10-10 10:00:00'), $article->getUpdated());
+
+        self::assertEquals(new ArticleAddedEvent($article->getId()), $article->shiftDomainEvent());
+        self::assertEquals(new ArticleUpdatedEvent($article->getId()), $article->shiftDomainEvent());
+        self::assertFalse($article->hasDomainEvents());
     }
 }