Snapshotter

composer.json

@@ -25,6 +25,7 @@
         "prooph/common": "4.0.x-dev",
         "prooph/event-store": "7.0.x-dev",
         "prooph/pdo-event-store": "dev-master",
+        "prooph/snapshot-store": "dev-master",
         "prooph/php-cs-fixer-config": "^0.1.1",
         "beberlei/assert": "^2.7"
     },
@@ -34,6 +35,10 @@
         "satooshi/php-coveralls": "^1.0",
         "malukenho/docheader": "^0.1.4"
     },
+    "suggest": {
+        "prooph/pdo-snapshot-store": "For PDO as snapshot store",
+        "prooph/mongodb-snapshot-store": "For MongoDB as snapshot store"
+    },
     "autoload": {
         "psr-4": {
             "Prooph\\Micro\\": "src/"

examples/Infrastructure/UserAggregateDefinition.php

@@ -18,6 +18,11 @@
 
 final class UserAggregateDefinition extends AbstractAggregateDefiniton
 {
+    public function aggregateType(): string
+    {
+        return 'user';
+    }
+
     public function streamName(string $aggregateId): StreamName
     {
         return new StreamName('user_stream'); // add aggregate id for one stream per aggregate

examples/Infrastructure/factories.php

@@ -23,6 +23,15 @@
 
         return $eventStore;
     },
+    'snapshotStore' => function (): \Prooph\SnapshotStore\SnapshotStore {
+        static $snapshotStore = null;
+
+        if (null === $snapshotStore) {
+            $snapshotStore = new \Prooph\SnapshotStore\InMemorySnapshotStore();
+        }
+
+        return $snapshotStore;
+    },
     'producer' => function (): callable {
         return function (Message $message): void {
         };

examples/Model/user_snapshotter.php

@@ -0,0 +1,13 @@
+<?php
+/**
+ * This file is part of the prooph/micro.
+ * (c) 2017-2017 prooph software GmbH <contact@prooph.de>
+ * (c) 2017-2017 Sascha-Oliver Prolic <saschaprolic@googlemail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace ProophExample\Micro;

examples/register_and_change_username.php

@@ -44,6 +44,7 @@
 
 $dispatch = \Prooph\Micro\Kernel\buildCommandDispatcher(
     $factories['eventStore'],
+    $factories['snapshotStore'],
     $commandMap,
     $factories['producer']
 );

examples/user_snapshotter.php

@@ -0,0 +1,47 @@
+<?php
+/**
+ * This file is part of the prooph/micro.
+ * (c) 2017-2017 prooph software GmbH <contact@prooph.de>
+ * (c) 2017-2017 Sascha-Oliver Prolic <saschaprolic@googlemail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace Prooph\MicroExample\Script;
+
+use Prooph\Common\Messaging\Message;
+use Prooph\EventStore\EventStore;
+use Prooph\Micro\SnapshotReadModel;
+use Prooph\MicroExample\Infrastructure\UserAggregateDefinition;
+
+$autoloader = require __DIR__ . '/../vendor/autoload.php';
+$autoloader->addPsr4('Prooph\\MicroExample\\', __DIR__);
+require 'Model/User.php';
+
+//We could also use a container here, if dependencies grow
+$factories = include 'Infrastructure/factories.php';
+
+$eventStore = $factories['eventStore']();
+
+/* @var EventStore $eventStore */
+
+$readModel = new SnapshotReadModel(
+    $eventStore,
+    $factories['snapshotStore'](),
+    new UserAggregateDefinition()
+);
+
+$projection = $eventStore->createReadModelProjection(
+    'user_snapshots',
+    $readModel
+);
+
+$projection
+    ->fromStream('user_stream')
+    ->whenAny(function ($state, Message $event): void {
+        $this->readModel()->stack('replay', $event);
+    })
+    ->run();

src/AbstractAggregateDefiniton.php

@@ -25,6 +25,11 @@ public function identifierName(): string
         return 'id';
     }
 
+    public function versionName(): string
+    {
+        return 'version';
+    }
+
     public function extractAggregateId(Message $message): string
     {
         $idProperty = $this->identifierName();
@@ -33,7 +38,7 @@ public function extractAggregateId(Message $message): string
 
         if (! array_key_exists($idProperty, $payload)) {
             throw new \RuntimeException(sprintf(
-                'Missing aggregate id %s in command payload of command %s. Payload was %s',
+                'Missing aggregate id %s in payload of message %s. Payload was %s',
                 $idProperty,
                 $message->messageName(),
                 json_encode($payload)

src/AggregateDefiniton.php

@@ -20,8 +20,12 @@
 
 interface AggregateDefiniton
 {
+    public function aggregateType(): string;
+
     public function identifierName(): string;
 
+    public function versionName(): string;
+
     public function extractAggregateId(Message $message): string;
 
     public function streamName(string $aggregateId): StreamName;

src/Kernel.php

@@ -20,14 +20,15 @@
 use Prooph\EventStore\StreamName;
 use Prooph\Micro\AggregateDefiniton;
 use Prooph\Micro\AggregateResult;
+use Prooph\SnapshotStore\SnapshotStore;
 
 const buildCommandDispatcher = 'Prooph\Micro\Kernel\buildCommandDispatcher';
 
 /**
  * builds a dispatcher to return a function that receives a messages and return the state
  *
  * usage:
- * $dispatch = buildDispatcher($eventStore, $commandMap, $producerFactory);
+ * $dispatch = buildDispatcher($eventStoreFactory, $snapshotStoreFactory, $commandMap, $producerFactory);
  * $state = $dispatch($message);
  *
  * $producerFactory is expected to be a callback that returns an instance of Prooph\ServiceBus\Async\MessageProducer.
@@ -48,13 +49,15 @@
  */
 function buildCommandDispatcher(
     callable $eventStoreFactory,
+    callable $snapshotStoreFactory,
     array $commandMap,
     callable $producerFactory,
     callable $startProducerTransaction = null,
     callable $commitProducerTransaction = null
 ): callable {
     return function (Message $message) use (
         $eventStoreFactory,
+        $snapshotStoreFactory,
         $commandMap,
         $producerFactory,
         $startProducerTransaction,
@@ -64,8 +67,8 @@ function buildCommandDispatcher(
             return getAggregateDefinition($message, $commandMap);
         };
 
-        $loadState = function (AggregateDefiniton $definiton) use ($message): array {
-            return loadState($message, $definiton);
+        $loadState = function (AggregateDefiniton $definiton) use ($message, $snapshotStoreFactory): array {
+            return loadState($snapshotStoreFactory(), $message, $definiton);
         };
 
         $loadEvents = function (array $state) use ($message, $getDefinition, $eventStoreFactory): Iterator {
@@ -145,9 +148,15 @@ function pipleline(callable $firstCallback, callable ...$callbacks): callable
 
 const loadState = 'Prooph\Micro\Kernel\loadState';
 
-function loadState(Message $message, AggregateDefiniton $definiton): array
+function loadState(SnapshotStore $snapshotStore, Message $message, AggregateDefiniton $definiton): array
 {
-    return []; // @todo: fetch from projections
+    $aggregate = $snapshotStore->get($definiton->aggregateType(), $definiton->extractAggregateId($message));
+
+    if (! $aggregate) {
+        return [];
+    }
+
+    return $aggregate->aggregateRoot();
 }
 
 const loadEvents = 'Prooph\Micro\Kernel\loadEvents';

src/SnapshotReadModel.php

@@ -0,0 +1,126 @@
+<?php
+/**
+ * This file is part of the prooph/micro.
+ * (c) 2017-2017 prooph software GmbH <contact@prooph.de>
+ * (c) 2017-2017 Sascha-Oliver Prolic <saschaprolic@googlemail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace Prooph\Micro;
+
+use Prooph\Common\Messaging\Message;
+use Prooph\EventStore\EventStore;
+use Prooph\EventStore\Projection\ReadModel;
+use Prooph\SnapshotStore\Snapshot;
+use Prooph\SnapshotStore\SnapshotStore;
+
+final class SnapshotReadModel implements ReadModel
+{
+    /**
+     * @var EventStore
+     */
+    private $eventStore;
+
+    /**
+     * @var SnapshotStore
+     */
+    private $snapshotStore;
+
+    /**
+     * @var AggregateDefiniton
+     */
+    private $aggregateDefinition;
+
+    /**
+     * @var array
+     */
+    private $cache = [];
+
+    public function __construct(
+        EventStore $eventStore,
+        SnapshotStore $snapshotStore,
+        AggregateDefiniton $aggregateDefiniton
+    ) {
+        $this->snapshotStore = $snapshotStore;
+        $this->aggregateDefinition = $aggregateDefiniton;
+    }
+
+    public function stack(string $operation, ...$events): void
+    {
+        foreach ($events as $event) {
+            if (! $event instanceof Message) {
+                throw new \RuntimeException(get_class($this) . ' can only handle events of type ' . Message::class);
+            }
+
+            $aggregateId = $this->aggregateDefinition->extractAggregateId($event);
+
+            if (! isset($this->cache[$aggregateId])) {
+                $snapshot = $this->snapshotStore->get(
+                    $this->aggregateDefinition->aggregateType(),
+                    $aggregateId
+                );
+
+                if (! $snapshot) {
+                    $state = [];
+                } else {
+                    $state = $snapshot->aggregateRoot();
+                }
+
+                $version = $state[$this->aggregateDefinition->versionName()] ?? 0;
+                ++$version;
+
+                $missingEvents = $this->eventStore->load(
+                    $this->aggregateDefinition->streamName($aggregateId),
+                    $version,
+                    null,
+                    $this->aggregateDefinition->metadataMatcher($aggregateId)
+                );
+
+                $state = $this->aggregateDefinition->apply($state, $missingEvents);
+            } else {
+                $state = $this->cache[$aggregateId];
+            }
+
+            $this->cache[$aggregateId] = $this->aggregateDefinition->apply($state, $event);
+        }
+    }
+
+    public function persist(): void
+    {
+        foreach ($this->cache as $aggregateId => $state) {
+            $this->snapshotStore->save(new Snapshot(
+                $this->aggregateDefinition->aggregateType(),
+                $aggregateId,
+                $state,
+                $state[$this->aggregateDefinition->versionName()],
+                new \DateTimeImmutable('now', new \DateTimeZone('UTC'))
+            ));
+        }
+
+        $this->cache = [];
+    }
+
+    public function init(): void
+    {
+        throw new \BadMethodCallException('Initializing a snapshot read model is not supported');
+    }
+
+    public function isInitialized(): bool
+    {
+        return true;
+    }
+
+    public function reset(): void
+    {
+        throw new \BadMethodCallException('Resetting a snapshot read model is not supported');
+    }
+
+    public function delete(): void
+    {
+        throw new \BadMethodCallException('Deleting a snapshot read model is not supported');
+    }
+}

tests/AbstractAggregateDefinitionTest.php

@@ -25,9 +25,10 @@ class AbstractAggregateDefinitionTest extends TestCase
     /**
      * @test
      */
-    public function it_returns_identifier(): void
+    public function it_returns_identifier_and_version_name(): void
     {
         $this->assertEquals('id', $this->createDefinition()->identifierName());
+        $this->assertEquals('version', $this->createDefinition()->versionName());
     }
 
     /**
@@ -116,6 +117,11 @@ public function streamName(string $aggregateId): StreamName
                 return new StreamName('foo');
             }
 
+            public function aggregateType(): string
+            {
+                return 'foo';
+            }
+
             public function apply(array $state, Message ...$events): array
             {
                 if (! isset($state['count'])) {