Timeline 👉 First MVP

.env

@@ -50,3 +50,7 @@ JWT_SECRET_KEY=%kernel.project_dir%/config/jwt/private.pem
 JWT_PUBLIC_KEY=%kernel.project_dir%/config/jwt/public.pem
 JWT_PASSPHRASE=925560e7309bd007e6827b45c2ff3166
 ###< lexik/jwt-authentication-bundle ###
+
+###> app ###
+ELASTICSEARCH_DSNS=127.0.0.1:9200
+###< app ###

.env.docker.dist

@@ -37,4 +37,16 @@ CORS_ALLOW_ORIGIN=^https?://(localhost|127\.0\.0\.1)(:[0-9]+)?$
 # MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages
 # MESSENGER_TRANSPORT_DSN=doctrine://default
 # MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages
+MESSENGER_TRANSPORT_BASE_DSN=amqp://guest:guest@rabbitmq:5672/%2f
+MESSENGER_TRANSPORT_DSN=amqp://guest:guest@rabbitmq:5672/%2f/messages
 ###< symfony/messenger ###
+
+###> lexik/jwt-authentication-bundle ###
+JWT_SECRET_KEY=%kernel.project_dir%/config/jwt/private.pem
+JWT_PUBLIC_KEY=%kernel.project_dir%/config/jwt/public.pem
+JWT_PASSPHRASE=925560e7309bd007e6827b45c2ff3166
+###< lexik/jwt-authentication-bundle ###
+
+###> app ###
+ELASTICSEARCH_DSNS=elasticsearch:9200
+###< app ###

config/packages/messenger.yaml

@@ -19,16 +19,12 @@ framework:
                 middleware:
                     - validation
         transports:
-            async_commands:
-                dsn: '%env(MESSENGER_TRANSPORT_BASE_DSN)%/commands'
-            sync_commands:
-                dsn: 'sync://'
-            events:
-                dsn: '%env(MESSENGER_TRANSPORT_BASE_DSN)%/events'
-            query:
-                dsn: 'sync://'
-            failed_messages:
-                dsn: '%env(MESSENGER_TRANSPORT_BASE_DSN)%/failed-messages'
+            async_commands: '%env(MESSENGER_TRANSPORT_BASE_DSN)%/commands'
+            sync_commands: 'sync://'
+            events: '%env(MESSENGER_TRANSPORT_BASE_DSN)%/events'
+            query: 'sync://'
+            failed_messages: '%env(MESSENGER_TRANSPORT_BASE_DSN)%/failed-messages'
+            projections: '%env(MESSENGER_TRANSPORT_BASE_DSN)%/projections'
         routing:
             # Consider Signing up an Author is a sync command.
             # With a proper UX, almost all the commands should
@@ -38,8 +34,7 @@ framework:
 
             # Domain Events are sent to asynchronous
             # transport to be processed later.
-            Cheeper\DomainModel\Follow\AuthorFollowed: events
-            Cheeper\DomainModel\Follow\AuthorUnfollowed: events
+            Cheeper\DomainModel\DomainEvent: [events, projections]
 
             # Queries are executed synchronously
             # in the same request that data is requested.

config/services.yaml

@@ -4,7 +4,8 @@
 # Put parameters here that don't need to change on each machine where the app is deployed
 # https://symfony.com/doc/current/best_practices/configuration.html#application-related-configuration
 parameters:
-
+    elasticsearch_config:
+        hosts: '%env(csv:ELASTICSEARCH_DSNS)%'
 services:
     # default configuration for services in *this* file
     _defaults:
@@ -94,3 +95,6 @@ services:
             - name: "doctrine.orm.entity_listener"
               event: "postPersist"
               entity: 'App\Entity\User'
+    Elasticsearch\Client:
+        factory: [ 'Elasticsearch\ClientBuilder', fromConfig ]
+        arguments: [ '%elasticsearch_config%' ]

docker-compose.yaml

@@ -6,9 +6,10 @@ services:
     ports:
       - "6379:6379"
   elasticsearch:
-    image: docker.elastic.co/elasticsearch/elasticsearch-oss:6.0.0
+    image: docker.elastic.co/elasticsearch/elasticsearch:7.15.1
     environment:
       ES_JAVA_OPTS: "-Xmx256m -Xms256m"
+      discovery.type: "single-node"
     ports:
       - "9200:9200"
       - "9300:9300"

src/App/API/DataProvider/TimelineDataProvider.php

@@ -0,0 +1,54 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\API\DataProvider;
+
+use ApiPlatform\Core\DataProvider\ItemDataProviderInterface;
+use ApiPlatform\Core\DataProvider\RestrictedDataProviderInterface;
+use App\API\Resources\Timeline;
+use Elasticsearch\Client as Elasticsearch;
+use Ramsey\Uuid\Uuid;
+
+final class TimelineDataProvider implements ItemDataProviderInterface, RestrictedDataProviderInterface
+{
+    public function __construct(
+        private Elasticsearch $elasticsearch
+    ) {
+    }
+
+    /** @inheritDoc */
+    public function getItem(string $resourceClass, $id, string $operationName = null, array $context = []): Timeline
+    {
+        $timeline = new Timeline();
+        $timeline->id = Uuid::fromString($id);
+        $timeline->cheeps = [];
+
+        $indexName = 'timelines_' . (string)$id;
+
+        if (!$this->elasticsearch->indices()->exists(['index' => $indexName])) {
+            return $timeline;
+        }
+
+        $result = $this->elasticsearch->search([
+            'index' => $indexName,
+            'body' => [
+                'sort' => [
+                    'cheep_date' => ['order' => 'desc']
+                ],
+                'query' => [
+                    'match_all' => new \stdClass()
+                ]
+            ]
+        ]);
+
+        $timeline->cheeps = $result['hits']['hits'];
+
+        return $timeline;
+    }
+
+    public function supports(string $resourceClass, string $operationName = null, array $context = []): bool
+    {
+        return Timeline::class === $resourceClass;
+    }
+}

src/App/API/Resources/Timeline.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\API\Resources;
+
+use ApiPlatform\Core\Annotation\ApiProperty;
+use ApiPlatform\Core\Annotation\ApiResource;
+use Ramsey\Uuid\UuidInterface;
+
+#[ApiResource(collectionOperations: [], itemOperations: ['get'])]
+final class Timeline
+{
+    #[ApiProperty(identifier: true)]
+    public ?UuidInterface $id = null;
+
+    public array $cheeps;
+}

src/Cheeper/Application/Projection/CheepProjection.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Cheeper\Application\Projection;
+
+use Cheeper\DomainModel\Cheep\CheepPosted;
+
+interface CheepProjection
+{
+    public function whenCheepPosted(CheepPosted $event): void;
+}

src/Cheeper/DomainModel/Cheep/Cheep.php

@@ -26,7 +26,7 @@ class Cheep
     public static function compose(AuthorId $authorId, CheepId $cheepId, CheepMessage $cheepMessage): self
     {
         $cheepDate = new CheepDate(
-            (new \DateTimeImmutable(timezone: new \DateTimeZone('UTC')))->format('Y-m-d')
+            (new \DateTimeImmutable(timezone: new \DateTimeZone('UTC')))->format('Y-m-d H:i:s')
         );
 
         return new self(

src/Cheeper/DomainModel/Cheep/CheepDate.php

@@ -19,12 +19,12 @@ final class CheepDate extends ValueObject
 
     public function date(): string
     {
-        return $this->date->format('Y-m-d');
+        return $this->date->format('Y-m-d H:i:s');
     }
 
     private function setDate(string $date): void
     {
-        $dateInstance = DateTimeImmutable::createFromFormat('Y-m-d', $date);
+        $dateInstance = DateTimeImmutable::createFromFormat('Y-m-d H:i:s', $date);
 
         if ($dateInstance === false) {
             throw new InvalidArgumentException("'$date' is not a valid datetime (Y-m-d formatted).");

src/Cheeper/DomainModel/Follow/Follows.php

@@ -12,5 +12,7 @@ interface Follows
     public function numberOfFollowersFor(AuthorId $authorId): int;
     public function add(Follow $follow): void;
     public function ofFromAuthorIdAndToAuthorId(AuthorId $fromAuthorId, AuthorId $toAuthorId): ?Follow;
+    /** @return Follow[] */
+    public function toAuthorId(AuthorId $authorId): array;
 }
 //end-snippet

src/Cheeper/Infrastructure/Persistence/DoctrineOrmFollows.php

@@ -38,5 +38,12 @@ final class DoctrineOrmFollows implements Follows
             'toAuthorId' => $toAuthorId,
         ]);
     }
+
+    public function toAuthorId(AuthorId $authorId): array
+    {
+        $repository = $this->em->getRepository(Follow::class);
+
+        return $repository->findBy(['toAuthorId' => $authorId]);
+    }
 }
 //end-snippet

src/Cheeper/Infrastructure/Persistence/InMemoryFollows.php

@@ -58,10 +58,11 @@ final class InMemoryFollows implements Follows
             select($this->collection, fn (Follow $u): bool => $u->fromAuthorId()->equals($fromAuthorId) && $u->toAuthorId()->equals($toAuthorId))
         );
 
-        if (null === $candidate) {
-            return null;
-        }
+        return $candidate ?? null;
+    }
 
-        return $candidate;
+    public function toAuthorId(AuthorId $authorId): array
+    {
+        return select($this->collection, fn (Follow $f): bool => $f->toAuthorId()->equals($authorId));
     }
 }

src/Cheeper/Infrastructure/Projection/SymfonyMessengerCheepProjectionToElasticsearch.php

@@ -0,0 +1,63 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Cheeper\Infrastructure\Projection;
+
+use App\Elasticsearch\Config;
+use Cheeper\Application\Projection\CheepProjection;
+use Cheeper\DomainModel\Author\AuthorId;
+use Cheeper\DomainModel\Author\Authors;
+use Cheeper\DomainModel\Cheep\CheepDoesNotExist;
+use Cheeper\DomainModel\Cheep\CheepId;
+use Cheeper\DomainModel\Cheep\CheepPosted;
+use Cheeper\DomainModel\Cheep\Cheeps;
+use Cheeper\DomainModel\Follow\Follows;
+use DateTimeImmutable;
+use DateTimeInterface;
+use Elasticsearch\Client as Elasticsearch;
+use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;
+
+final class SymfonyMessengerCheepProjectionToElasticsearch implements CheepProjection, MessageSubscriberInterface
+{
+    public function __construct(
+        private Elasticsearch $elasticsearch,
+        private Follows $follows,
+        private Cheeps $cheeps,
+    ) {
+    }
+
+    public function whenCheepPosted(CheepPosted $event): void
+    {
+        $cheepId = CheepId::fromString($event->cheepId());
+
+        $cheep = $this->cheeps->ofId($cheepId);
+
+        if (null === $cheep) {
+            throw CheepDoesNotExist::withIdOf($cheepId);
+        }
+
+        $authorId = $cheep->authorId();
+        $follows = $this->follows->toAuthorId($authorId);
+
+        foreach ($follows as $follow) {
+            $this->elasticsearch->index([
+                'index' => sprintf("timelines_%s", $follow->fromAuthorId()->toString()),
+                'id' => $cheepId->toString(),
+                'body' => [
+                    'cheep_id' => $cheepId->toString(),
+                    'cheep_message' => $cheep->cheepMessage()->message(),
+                    'cheep_date' => DateTimeImmutable::createFromFormat('Y-m-d H:i:s', $cheep->cheepDate()->date())->setTimezone(new \DateTimeZone('UTC'))->format(DateTimeInterface::ATOM)
+                ]
+            ]);
+        }
+    }
+
+    public static function getHandledMessages(): iterable
+    {
+        yield CheepPosted::class => [
+            'method' => 'whenCheepPosted',
+            'from_transport' => 'projections'
+        ];
+    }
+}