WIP: FEATURE: Course read model example

README.md

@@ -48,7 +48,7 @@ And you should get ...no output at all. That's because the example script curren
 Try changing the script to test, that the business rules are actually enforced, for example you could add the line:
 
 ```php
-$commandHandler->handle(new SubscribeStudentToCourse(CourseId::fromString('c1'), StudentId::fromString('s2')));
+$app->handle(new SubscribeStudentToCourse(CourseId::fromString('c1'), StudentId::fromString('s2')));
 ```
 
 to the end of the file, which should lead to the following exception:

composer.json

@@ -33,17 +33,18 @@
     "ext-ctype": "*",
     "ramsey/uuid": "^4.7",
     "webmozart/assert": "^1.11",
-    "wwwision/dcb-eventstore": "@dev",
-    "wwwision/dcb-eventstore-doctrine": "@dev",
-    "wwwision/dcb-library": "@dev"
+    "wwwision/dcb-eventstore": "dev-main as 3.0.0",
+    "wwwision/types": "^1.1",
+    "wwwision/dcb-eventstore-doctrine": "dev-main as 3.0.0",
+    "wwwision/dcb-library": "@dev",
+    "wwwision/dcb-library-doctrine": "@dev"
   },
   "require-dev": {
     "roave/security-advisories": "dev-latest",
     "phpstan/phpstan": "^1.10",
     "squizlabs/php_codesniffer": "^4.0.x-dev",
     "phpunit/phpunit": "^10.2",
     "behat/behat": "^3.13",
-    "wwwision/dcb-library-phpstanextension": "@dev",
     "phpstan/extension-installer": "^1.3"
   },
   "replace": {

index.php

@@ -2,8 +2,6 @@
 declare(strict_types=1);
 
 use Doctrine\DBAL\DriverManager;
-use Wwwision\DCBEventStore\EventStore;
-use Wwwision\DCBEventStoreDoctrine\DoctrineEventStore;
 use Wwwision\DCBExample\App;
 use Wwwision\DCBExample\Commands\Command;
 use Wwwision\DCBExample\Commands\CreateCourse;
@@ -16,36 +14,33 @@
 use Wwwision\DCBExample\Types\CourseId;
 use Wwwision\DCBExample\Types\CourseTitle;
 use Wwwision\DCBExample\Types\StudentId;
+use Wwwision\DCBLibraryDoctrine\DbalCheckpointStorage;
 
 require __DIR__ . '/vendor/autoload.php';
 
-/** We use an in-memory SQLite database for the events (@see https://www.doctrine-project.org/projects/doctrine-dbal/en/2.4/reference/configuration.html for how to configure other database backends) **/
-$connection = DriverManager::getConnection(['url' => 'sqlite:///:memory:']);
-
-/** The second parameter is the table name to store the events in **/
-$eventStore = DoctrineEventStore::create($connection, 'dcb_events');
-
-/** The {@see EventStore::setup()} method is used to make sure that the Events Store backend is set up (i.e. required tables are created and their schema up-to-date) **/
-$eventStore->setup();
+$dsn = $argv[1] ?? 'sqlite:///:memory:';
+$connection = DriverManager::getConnection(['url' => $dsn]);
 
 /** @var {@see App} is the central authority to handle {@see Command}s */
-$commandHandler = new App($eventStore);
+$app = new App($connection);
 
 // Example:
 // 1. Create a course (c1)
-$commandHandler->handle(new CreateCourse(CourseId::fromString('c1'), CourseCapacity::fromInteger(10), CourseTitle::fromString('Course 02')));
+$app->handle(new CreateCourse(CourseId::fromString('c1'), CourseCapacity::fromInteger(10), CourseTitle::fromString('Course 01')));
+$app->handle(new CreateCourse(CourseId::fromString('c2'), CourseCapacity::fromInteger(15), CourseTitle::fromString('Course 02')));
 
 // 2. rename it, register a student (s1) and subscribe it to the course, change the course capacity, unregister the student
-$commandHandler->handle(new RenameCourse(CourseId::fromString('c1'), CourseTitle::fromString('Course 01 renamed again')));
+$app->handle(new RenameCourse(CourseId::fromString('c1'), CourseTitle::fromString('Course 01 renamed again')));
 
 // 3. register a student (s1) in the system
-$commandHandler->handle(new RegisterStudent(StudentId::fromString('s1')));
+$app->handle(new RegisterStudent(StudentId::fromString('s1')));
 
 // 4. subscribe student (s1) to course (s1)
-$commandHandler->handle(new SubscribeStudentToCourse(CourseId::fromString('c1'), StudentId::fromString('s1')));
+$app->handle(new SubscribeStudentToCourse(CourseId::fromString('c1'), StudentId::fromString('s1')));
+$app->handle(new SubscribeStudentToCourse(CourseId::fromString('c2'), StudentId::fromString('s1')));
 
 // 5. change capacity of course (c1) to 5
-$commandHandler->handle(new UpdateCourseCapacity(CourseId::fromString('c1'), CourseCapacity::fromInteger(5)));
+$app->handle(new UpdateCourseCapacity(CourseId::fromString('c1'), CourseCapacity::fromInteger(5)));
 
 // 6. unsubscribe student (s1) from course (c1)
-$commandHandler->handle(new UnsubscribeStudentFromCourse(CourseId::fromString('c1'), StudentId::fromString('s1')));
+$app->handle(new UnsubscribeStudentFromCourse(CourseId::fromString('c1'), StudentId::fromString('s1')));

src/Adapters/DbalCourseProjectionAdapter.php

@@ -0,0 +1,123 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Wwwision\DCBExample\Adapters;
+
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Schema\Comparator;
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\DBAL\Types\Types;
+use Webmozart\Assert\Assert;
+use Wwwision\DCBExample\ReadModel\Course\Course;
+use Wwwision\DCBExample\ReadModel\Course\CourseProjectionAdapter;
+use Wwwision\DCBExample\ReadModel\Course\Courses;
+use Wwwision\DCBExample\Types\CourseId;
+use Wwwision\DCBExample\Types\CourseTitle;
+use Wwwision\Types\Parser;
+use Wwwision\Types\Schema\StringSchema;
+use function Wwwision\Types\instantiate;
+
+final readonly class DbalCourseProjectionAdapter implements CourseProjectionAdapter
+{
+
+    private const TABLE_NAME = 'dcv_example_courses_p_courses';
+
+    public function __construct(
+        private Connection $connection,
+    ) {
+    }
+
+    public function saveCourse(Course $course): void
+    {
+        $data = self::courseToDatabaseRow($course);
+        $assignments = [];
+        $parameters = [];
+        foreach ($data as $columnName => $value) {
+            $assignments[$columnName] = $this->connection->quoteIdentifier($columnName) . ' = :' . $columnName;
+            $parameters[$columnName] = $value;
+        }
+        $sql = 'INSERT INTO ' . self::TABLE_NAME . ' SET ' . (implode(', ', $assignments)) . ' ON DUPLICATE KEY UPDATE ' . (implode(', ', $assignments));
+        $this->connection->executeStatement($sql, $parameters);
+    }
+
+    public function courses(): Courses
+    {
+        $rows = $this->connection->fetchAllAssociative('SELECT * FROM ' . self::TABLE_NAME);
+        $instances = array_map(self::courseFromDatabaseRow(...), $rows);
+        return instantiate(Courses::class, $instances);
+    }
+
+    public function courseById(CourseId $courseId): ?Course
+    {
+        $row = $this->connection->fetchAssociative('SELECT * FROM ' . self::TABLE_NAME . ' WHERE id = :courseId', ['courseId' => $courseId->value]);
+        if ($row === false) {
+            return null;
+        }
+        return self::courseFromDatabaseRow($row);
+    }
+
+    // -------- HELPERS, INFRASTRUCTURE ----------
+
+    public function setup(): void
+    {
+        $schemaManager = $this->connection->createSchemaManager();
+        $schemaDiff = (new Comparator())->compareSchemas($schemaManager->introspectSchema(), self::databaseSchema());
+        foreach ($schemaDiff->toSaveSql($this->connection->getDatabasePlatform()) as $statement) {
+            $this->connection->executeStatement($statement);
+        }
+    }
+
+    public function reset(): void
+    {
+        $this->connection->executeStatement('TRUNCATE TABLE ' . self::TABLE_NAME);
+    }
+
+    private static function databaseSchema(): Schema
+    {
+        $schema = new Schema();
+        $table = $schema->createTable(self::TABLE_NAME);
+
+        $table->addColumn('id', Types::STRING, ['length' => self::maxLength(CourseId::class)]);
+        $table->addColumn('title', Types::STRING, ['length' => self::maxLength(CourseTitle::class), 'notnull' => false]);
+        $table->addColumn('state', Types::JSON);
+        $table->setPrimaryKey(['id']);
+
+        return $schema;
+    }
+
+    /**
+     * @param class-string $className
+     */
+    private static function maxLength(string $className): int
+    {
+        $schema = Parser::getSchema($className);
+        Assert::isInstanceOf($schema, StringSchema::class, sprintf('Failed to determine max length for class %s: Expected an instance of %%2$s. Got: %%s', $className));
+        Assert::notNull($schema->maxLength, sprintf('Failed to determine max length for class %s: No maxLength constraint defined', $className));
+        return $schema->maxLength;
+    }
+
+    /**
+     * @param array<mixed> $row
+     */
+    private static function courseFromDatabaseRow(array $row): Course
+    {
+        return instantiate(Course::class, [
+            'id' => $row['id'],
+            'title' => $row['title'],
+            'state' => json_decode($row['state'], true, 512, JSON_THROW_ON_ERROR),
+        ]);
+    }
+
+    /**
+     * @return array<mixed>
+     */
+    private static function courseToDatabaseRow(Course $course): array
+    {
+        return [
+            'id' => $course->id->value,
+            'title' => $course->title->value,
+            'state' => json_encode($course->state, JSON_THROW_ON_ERROR),
+        ];
+    }
+}

src/App.php

@@ -4,9 +4,12 @@
 
 namespace Wwwision\DCBExample;
 
+use Doctrine\DBAL\Connection;
 use RuntimeException;
 use Wwwision\DCBEventStore\EventStore;
 use Wwwision\DCBEventStore\Types\Tags;
+use Wwwision\DCBEventStoreDoctrine\DoctrineEventStore;
+use Wwwision\DCBExample\Adapters\DbalCourseProjectionAdapter;
 use Wwwision\DCBExample\Commands\Command;
 use Wwwision\DCBExample\Commands\CreateCourse;
 use Wwwision\DCBExample\Commands\RegisterStudent;
@@ -20,6 +23,7 @@
 use Wwwision\DCBExample\Events\StudentRegistered;
 use Wwwision\DCBExample\Events\StudentSubscribedToCourse;
 use Wwwision\DCBExample\Events\StudentUnsubscribedFromCourse;
+use Wwwision\DCBExample\ReadModel\Course\CourseProjection;
 use Wwwision\DCBExample\Types\CourseId;
 use Wwwision\DCBExample\Types\CourseIds;
 use Wwwision\DCBExample\Types\CourseState;
@@ -28,11 +32,13 @@
 use Wwwision\DCBExample\Types\StudentId;
 use Wwwision\DCBLibrary\Adapters\SynchronousCatchUpQueue;
 use Wwwision\DCBLibrary\EventHandling\EventHandlers;
+use Wwwision\DCBLibrary\EventHandling\ProjectionEventHandler;
 use Wwwision\DCBLibrary\EventPublisher;
 use Wwwision\DCBLibrary\Exceptions\ConstraintException;
 use Wwwision\DCBLibrary\Projection\CompositeProjection;
 use Wwwision\DCBLibrary\Projection\InMemoryProjection;
 use Wwwision\DCBLibrary\Projection\Projection;
+use Wwwision\DCBLibraryDoctrine\DbalCheckpointStorage;
 use function sprintf;
 
 /**
@@ -42,9 +48,31 @@
 {
     private EventPublisher $eventPublisher;
 
-    public function __construct(EventStore $eventStore)
+    public function __construct(Connection $connection)
     {
-        $this->eventPublisher = new EventPublisher($eventStore, new EventSerializer(), new SynchronousCatchUpQueue($eventStore, EventHandlers::create()));
+
+        /** The second parameter is the table name to store the events in **/
+        $eventStore = DoctrineEventStore::create($connection, 'dcb_events');
+
+        /** The {@see EventStore::setup()} method is used to make sure that the Events Store backend is set up (i.e. required tables are created and their schema up-to-date) **/
+        $eventStore->setup();
+        $connection->executeStatement('TRUNCATE TABLE dcb_events');
+
+        $eventSerializer = new EventSerializer();
+
+        $courseProjection = new CourseProjection(new DbalCourseProjectionAdapter($connection));
+        $courseProjection->setup();
+        $courseProjection->reset();
+
+        $courseProjectionEventHandler = new ProjectionEventHandler($courseProjection, new DbalCheckpointStorage($connection, 'dcb_checkpoints', $courseProjection::class), $eventSerializer);
+        $courseProjectionEventHandler->setup();
+        $courseProjectionEventHandler->reset();
+
+        $eventHandlers = EventHandlers::create()
+            ->with('CourseProjection', $courseProjectionEventHandler);
+
+
+        $this->eventPublisher = new EventPublisher($eventStore, $eventSerializer, new SynchronousCatchUpQueue($eventStore, $eventHandlers));
     }
 
     public function handle(Command $command): void

src/CourseStateProjection.php

@@ -1,4 +1,4 @@
-<?php /** @noinspection ALL */
+<?php /** @noinspection PhpUnusedPrivateMethodInspection */
 
 declare(strict_types=1);
 
@@ -52,17 +52,17 @@ public function apply($state, DomainEvent $domainEvent, EventEnvelope $eventEnve
         return $state;
     }
 
-    private function whenCourseCreated(CourseState $state, CourseCreated $event): CourseState
+    private function whenCourseCreated(CourseState $state, CourseCreated $domainEvent): CourseState
     {
-        return $state->withValue(CourseStateValue::CREATED)->withCapacity($event->initialCapacity);
+        return $state->withValue(CourseStateValue::CREATED)->withCapacity($domainEvent->initialCapacity);
     }
 
-    private function whenCourseCapacityChanged(CourseState $state, CourseCapacityChanged $event): CourseState
+    private function whenCourseCapacityChanged(CourseState $state, CourseCapacityChanged $domainEvent): CourseState
     {
-        return $state->withCapacity($event->newCapacity);
+        return $state->withCapacity($domainEvent->newCapacity);
     }
 
-    private function whenStudentSubscribedToCourse(CourseState $state, StudentSubscribedToCourse $event): CourseState
+    private function whenStudentSubscribedToCourse(CourseState $state, StudentSubscribedToCourse $domainEvent): CourseState
     {
         $state = $state->withNumberOfSubscriptions($state->numberOfSubscriptions + 1);
         if ($state->numberOfSubscriptions === $state->capacity->value) {
@@ -71,7 +71,7 @@ private function whenStudentSubscribedToCourse(CourseState $state, StudentSubscr
         return $state;
     }
 
-    private function whenStudentUnsubscribedFromCourse(CourseState $state, StudentUnsubscribedFromCourse $event): CourseState
+    private function whenStudentUnsubscribedFromCourse(CourseState $state, StudentUnsubscribedFromCourse $domainEvent): CourseState
     {
         $state = $state->withNumberOfSubscriptions($state->numberOfSubscriptions + 1);
         if ($state->numberOfSubscriptions < $state->capacity->value && $state->value === CourseStateValue::FULLY_BOOKED) {

src/EventSerializer.php

@@ -9,8 +9,7 @@
 use Webmozart\Assert\Assert;
 use Wwwision\DCBEventStore\Types\Event;
 use Wwwision\DCBEventStore\Types\EventData;
-use Wwwision\DCBExample\Events\DomainEvent;
-use Wwwision\DCBLibrary\DomainEvent as BaseDomainEvent;
+use Wwwision\DCBLibrary\DomainEvent;
 use Wwwision\DCBEventStore\Types\EventId;
 use Wwwision\DCBEventStore\Types\EventMetadata;
 use Wwwision\DCBEventStore\Types\EventType;
@@ -22,6 +21,7 @@
 use function strrpos;
 use function substr;
 
+use function Wwwision\Types\instantiate;
 use const JSON_THROW_ON_ERROR;
 
 /**
@@ -39,12 +39,12 @@ public function convertEvent(Event $event): DomainEvent
         Assert::isArray($payload);
         /** @var class-string<DomainEvent> $eventClassName */
         $eventClassName = '\\Wwwision\\DCBExample\\Events\\' . $event->type->value;
-        $domainEvent = $eventClassName::fromArray($payload);
+        $domainEvent = instantiate($eventClassName, $payload);
         Assert::isInstanceOf($domainEvent, DomainEvent::class);
         return $domainEvent;
     }
 
-    public function convertDomainEvent(BaseDomainEvent $domainEvent): Event
+    public function convertDomainEvent(DomainEvent $domainEvent): Event
     {
         try {
             $eventData = json_encode($domainEvent, JSON_THROW_ON_ERROR);

src/Events/CourseCapacityChanged.php

@@ -4,10 +4,10 @@
 
 namespace Wwwision\DCBExample\Events;
 
-use Webmozart\Assert\Assert;
 use Wwwision\DCBEventStore\Types\Tags;
 use Wwwision\DCBExample\Types\CourseCapacity;
 use Wwwision\DCBExample\Types\CourseId;
+use Wwwision\DCBLibrary\DomainEvent;
 
 /**
  * Domain Events that occurs when the total capacity of a course has changed
@@ -20,21 +20,6 @@ public function __construct(
     ) {
     }
 
-    /**
-     * @param array<string, mixed> $data
-     */
-    public static function fromArray(array $data): self
-    {
-        Assert::keyExists($data, 'courseId');
-        Assert::string($data['courseId']);
-        Assert::keyExists($data, 'newCapacity');
-        Assert::numeric($data['newCapacity']);
-        return new self(
-            CourseId::fromString($data['courseId']),
-            CourseCapacity::fromInteger((int)$data['newCapacity']),
-        );
-    }
-
     public function tags(): Tags
     {
         return Tags::create($this->courseId->toTag());