[POC] Replaced document-manager by event-sourcing and CQRS

Builder/ArticleIndexBuilder.php

@@ -69,6 +69,6 @@ private function buildForManager(Manager $manager, $destroy)
         }
 
         $this->output->writeln(sprintf('Drop and create index for "<comment>%s</comment>" manager.', $name));
-        $manager->dropAndCreateIndex();
+        // $manager->dropAndCreateIndex();
     }
 }

Controller/ArticleController.php

@@ -23,15 +23,45 @@
 use ONGR\ElasticsearchDSL\Query\TermLevel\RangeQuery;
 use ONGR\ElasticsearchDSL\Query\TermLevel\TermQuery;
 use ONGR\ElasticsearchDSL\Sort\FieldSort;
+use Prooph\Common\Event\ProophActionEventEmitter;
+use Prooph\Common\Messaging\FQCNMessageFactory;
+use Prooph\EventStore\ActionEventEmitterEventStore;
+use Prooph\EventStore\Pdo\MySqlEventStore;
+use Prooph\EventStore\Pdo\PersistenceStrategy\MySqlAggregateStreamStrategy;
+use Prooph\EventStore\Pdo\Projection\MySqlProjectionManager;
+use Prooph\EventStoreBusBridge\EventPublisher;
+use Prooph\ServiceBus\CommandBus;
+use Prooph\ServiceBus\EventBus;
+use Prooph\ServiceBus\Plugin\Router\CommandRouter;
+use Prooph\ServiceBus\Plugin\Router\EventRouter;
+use Prooph\SnapshotStore\Pdo\PdoSnapshotStore;
+use Ramsey\Uuid\Uuid;
 use Sulu\Bundle\ArticleBundle\Admin\ArticleAdmin;
 use Sulu\Bundle\ArticleBundle\Document\ArticleDocument;
 use Sulu\Bundle\ArticleBundle\ListBuilder\ElasticSearchFieldDescriptor;
 use Sulu\Bundle\ArticleBundle\Metadata\ArticleViewDocumentIdTrait;
-use Sulu\Component\Content\Form\Exception\InvalidFormException;
+use Sulu\Bundle\ArticleBundle\Prooph\Infrastruture\ArticleRepository;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Article;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Command\DeleteArticleCommand;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Command\PostArticleCommand;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Command\PublishArticleCommand;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Command\PutArticleCommand;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Command\UnpublishArticleCommand;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Event\CreateTranslation;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Event\ModifyTranslationStructure;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Event\PublishTranslation;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Event\RemoveArticle;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Event\UnpublishTranslation;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Handler\DeleteArticleHandler;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Handler\PostArticleHandler;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Handler\PublishArticleHandler;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Handler\PutArticleHandler;
+use Sulu\Bundle\ArticleBundle\Prooph\Model\Handler\UnpublishArticleHandler;
+use Sulu\Bundle\ArticleBundle\Prooph\Projection\ArticleDocumentProjector;
+use Sulu\Bundle\ArticleBundle\Prooph\Projection\ArticleIndexProjector;
 use Sulu\Component\Content\Mapper\ContentMapperInterface;
 use Sulu\Component\DocumentManager\DocumentManagerInterface;
 use Sulu\Component\DocumentManager\Metadata\BaseMetadataFactory;
-use Sulu\Component\Rest\Exception\MissingParameterException;
 use Sulu\Component\Rest\Exception\RestException;
 use Sulu\Component\Rest\ListBuilder\ListRepresentation;
 use Sulu\Component\Rest\RequestParametersTrait;
@@ -279,14 +309,23 @@ public function getAction($uuid, Request $request)
      */
     public function postAction(Request $request)
     {
-        $action = $request->get('action');
-        $document = $this->getDocumentManager()->create(self::DOCUMENT_TYPE);
+        $id = Uuid::uuid4()->toString();
+
         $locale = $this->getRequestParameter($request, 'locale', true);
-        $data = $request->request->all();
+        $this->getCommandBus()->dispatch(
+            new PostArticleCommand(
+                [
+                    'id' => $id,
+                    'locale' => $locale,
+                    'userId' => $this->getUser()->getId(),
+                    'requestData' => $request->request->all(),
+                ]
+            )
+        );
 
-        $this->persistDocument($data, $document, $locale);
-        $this->handleActionParameter($action, $document, $locale);
-        $this->getDocumentManager()->flush();
+        $this->handleActionParameter($id, $request->get('action'), $locale);
+
+        $document = $this->getDocumentManager()->find($id, $locale);
 
         return $this->handleView(
             $this->view($document)->setSerializationContext(
@@ -309,7 +348,6 @@ public function putAction(Request $request, $uuid)
     {
         $locale = $this->getRequestParameter($request, 'locale', true);
         $action = $request->get('action');
-        $data = $request->request->all();
 
         $document = $this->getDocumentManager()->find(
             $uuid,
@@ -322,9 +360,20 @@ public function putAction(Request $request, $uuid)
 
         $this->get('sulu_hash.request_hash_checker')->checkHash($request, $document, $document->getUuid());
 
-        $this->persistDocument($data, $document, $locale);
-        $this->handleActionParameter($action, $document, $locale);
-        $this->getDocumentManager()->flush();
+        $this->getCommandBus()->dispatch(
+            new PutArticleCommand(
+                [
+                    'id' => $uuid,
+                    'locale' => $locale,
+                    'userId' => $this->getUser()->getId(),
+                    'requestData' => $request->request->all(),
+                ]
+            )
+        );
+
+        $this->handleActionParameter($uuid, $action, $locale);
+
+        $document = $this->getDocumentManager()->find($uuid, $locale);
 
         return $this->handleView(
             $this->view($document)->setSerializationContext(
@@ -346,11 +395,15 @@ public function cdeleteAction(Request $request)
     {
         $ids = array_filter(explode(',', $request->get('ids', '')));
 
-        $documentManager = $this->getDocumentManager();
         foreach ($ids as $id) {
-            $document = $documentManager->find($id);
-            $documentManager->remove($document);
-            $documentManager->flush();
+            $this->getCommandBus()->dispatch(
+                new DeleteArticleCommand(
+                    [
+                        'id' => $id,
+                        'userId' => $this->getUser()->getId(),
+                    ]
+                )
+            );
         }
 
         return $this->handleView($this->view(null));
@@ -365,10 +418,14 @@ public function cdeleteAction(Request $request)
      */
     public function deleteAction($id)
     {
-        $documentManager = $this->getDocumentManager();
-        $document = $documentManager->find($id);
-        $documentManager->remove($document);
-        $documentManager->flush();
+        $this->getCommandBus()->dispatch(
+            new DeleteArticleCommand(
+                [
+                    'id' => $id,
+                    'userId' => $this->getUser()->getId(),
+                ]
+            )
+        );
 
         return $this->handleView($this->view(null));
     }
@@ -378,7 +435,7 @@ public function deleteAction($id)
      *
      * @Post("/articles/{uuid}")
      *
-     * @param string  $uuid
+     * @param string $uuid
      * @param Request $request
      *
      * @return Response
@@ -397,9 +454,15 @@ public function postTriggerAction($uuid, Request $request)
         try {
             switch ($action) {
                 case 'unpublish':
-                    $document = $this->getDocumentManager()->find($uuid, $locale);
-                    $this->getDocumentManager()->unpublish($document, $locale);
-                    $this->getDocumentManager()->flush();
+                    $this->getCommandBus()->dispatch(
+                        new UnpublishArticleCommand(
+                            [
+                                'id' => $uuid,
+                                'locale' => $locale,
+                                'userId' => $this->getUser()->getId(),
+                            ]
+                        )
+                    );
 
                     $data = $this->getDocumentManager()->find($uuid, $locale);
 
@@ -486,47 +549,6 @@ public function getSecurityContext()
         return ArticleAdmin::SECURITY_CONTEXT;
     }
 
-    /**
-     * Persists the document using the given information.
-     *
-     * @param array  $data
-     * @param object $document
-     * @param string $locale
-     *
-     * @throws InvalidFormException
-     * @throws MissingParameterException
-     */
-    private function persistDocument($data, $document, $locale)
-    {
-        $formType = $this->getMetadataFactory()->getMetadataForAlias('article')->getFormType();
-        $form = $this->createForm(
-            $formType,
-            $document,
-            [
-                // disable csrf protection, since we can't produce a token, because the form is cached on the client
-                'csrf_protection' => false,
-            ]
-        );
-        $form->submit($data, false);
-
-        if (!$form->isValid()) {
-            throw new InvalidFormException($form);
-        }
-
-        if (array_key_exists('author', $data) && null === $data['author']) {
-            $document->setAuthor(null);
-        }
-
-        $this->getDocumentManager()->persist(
-            $document,
-            $locale,
-            [
-                'user' => $this->getUser()->getId(),
-                'clear_missing_content' => false,
-            ]
-        );
-    }
-
     /**
      * Returns document-manager.
      *
@@ -545,20 +567,18 @@ protected function getMapper()
         return $this->get('sulu.content.mapper');
     }
 
-    /**
-     * Delegates actions by given actionParameter, which can be retrieved from the request.
-     *
-     * @param string $actionParameter
-     * @param object $document
-     * @param string $locale
-     */
-    private function handleActionParameter($actionParameter, $document, $locale)
+    private function handleActionParameter(string $id, ?string $action, string $locale)
     {
-        switch ($actionParameter) {
-            case 'publish':
-                $this->getDocumentManager()->publish($document, $locale);
-
-                break;
+        if ($action === 'publish') {
+            $this->getCommandBus()->dispatch(
+                new PublishArticleCommand(
+                    [
+                        'id' => $id,
+                        'locale' => $locale,
+                        'userId' => $this->getUser()->getId(),
+                    ]
+                )
+            );
         }
     }
 
@@ -604,4 +624,61 @@ protected function getMetadataFactory()
     {
         return $this->get('sulu_document_manager.metadata_factory.base');
     }
+
+    protected function getCommandBus()
+    {
+        $pdo = new \PDO('mysql:dbname=su_article_prooph;host=127.0.0.1', 'root', '');
+        $eventStore = new MySqlEventStore(new FQCNMessageFactory(), $pdo, new MySqlAggregateStreamStrategy());
+        $eventEmitter = new ProophActionEventEmitter();
+        $eventStore = new ActionEventEmitterEventStore($eventStore, $eventEmitter);
+
+        $eventBus = new EventBus($eventEmitter);
+        $eventPublisher = new EventPublisher($eventBus);
+        $eventPublisher->attachToEventStore($eventStore);
+
+        $pdoSnapshotStore = new PdoSnapshotStore($pdo);
+        $userRepository = new ArticleRepository(Article::class, $eventStore, $pdoSnapshotStore);
+
+        $projectionManager = new MySqlProjectionManager($eventStore, $pdo);
+
+        $structureFactory = $this->get('sulu_content.structure.factory');
+        $routeGenerator = $this->get('sulu_route.chain_generator');
+        $conflictResolver = $this->get('sulu_route.manager.conflict_resolver.auto_increment');
+
+        $commandBus = new CommandBus();
+        $router = new CommandRouter();
+        $router->route(PostArticleCommand::class)->to(new PostArticleHandler($userRepository, $structureFactory, $routeGenerator, $conflictResolver));
+        $router->route(PublishArticleCommand::class)->to(new PublishArticleHandler($userRepository));
+        $router->route(UnpublishArticleCommand::class)->to(new UnpublishArticleHandler($userRepository));
+        $router->route(PutArticleCommand::class)->to(new PutArticleHandler($userRepository, $structureFactory));
+        $router->route(DeleteArticleCommand::class)->to(new DeleteArticleHandler($userRepository));
+        $router->attachToMessageBus($commandBus);
+
+        $documentProjector = new ArticleDocumentProjector(
+            $this->getDocumentManager(),
+            $this->getMetadataFactory(),
+            $this->get('form.factory')
+        );
+
+        $indexProjector = new ArticleIndexProjector(
+            $this->getDocumentManager(),
+            $this->get('sulu_article.elastic_search.article_indexer'),
+            $this->get('sulu_article.elastic_search.article_live_indexer')
+        );
+
+        $eventRouter = new EventRouter();
+        $eventRouter->route(CreateTranslation::class)->to([$documentProjector, 'onCreateTranslation']);
+        $eventRouter->route(CreateTranslation::class)->to([$indexProjector, 'onCreateTranslation']);
+        $eventRouter->route(PublishTranslation::class)->to([$documentProjector, 'onPublishTranslation']);
+        $eventRouter->route(PublishTranslation::class)->to([$indexProjector, 'onPublishTranslation']);
+        $eventRouter->route(UnpublishTranslation::class)->to([$documentProjector, 'onUnpublishTranslation']);
+        $eventRouter->route(UnpublishTranslation::class)->to([$indexProjector, 'onUnpublishTranslation']);
+        $eventRouter->route(ModifyTranslationStructure::class)->to([$documentProjector, 'onModifyTranslationStructure']);
+        $eventRouter->route(ModifyTranslationStructure::class)->to([$indexProjector, 'onModifyTranslationStructure']);
+        $eventRouter->route(RemoveArticle::class)->to([$documentProjector, 'onRemoveArticle']);
+        $eventRouter->route(RemoveArticle::class)->to([$indexProjector, 'onRemoveArticle']);
+        $eventRouter->attachToMessageBus($eventBus);
+
+        return $commandBus;
+    }
 }

Document/Index/ArticleIndexer.php

@@ -365,11 +365,11 @@ protected function removeArticle($id)
     /**
      * {@inheritdoc}
      */
-    public function remove($document)
+    public function remove($uuid)
     {
         $repository = $this->manager->getRepository($this->documentFactory->getClass('article'));
         $search = $repository->createSearch()
-            ->addQuery(new TermQuery('uuid', $document->getUuid()))
+            ->addQuery(new TermQuery('uuid', $uuid))
             ->setSize(1000);
         foreach ($repository->findDocuments($search) as $viewDocument) {
             $this->manager->remove($viewDocument);

Document/Index/IndexerInterface.php

@@ -44,9 +44,9 @@ public function index(ArticleDocument $document);
     /**
      * Removes document from index.
      *
-     * @param ArticleDocument $document
+     * @param string $uuid
      */
-    public function remove($document);
+    public function remove($uuid);
 
     /**
      * Flushes index.