Merge pull request #7 from PcComponentes/GG-403-add-get-filtered-event-to-repository

Added getEventsFilteredByAggregate to EventStoreRepository

Author: jgmullor

src/Repository/EventStoreRepository.php

@@ -11,10 +11,16 @@ interface EventStoreRepository extends DddEventStoreRepository
 {
     public function countEventsFor(Uuid $aggregateId): int;
 
+    public function countGivenEventsByAggregate(Uuid $aggregateId, string ...$events): int;
+
+    public function countEventsFilteredByAggregate(Uuid $aggregateId, string ...$events): int;
+
     public function countEventsForSince(Uuid $aggregateId, DateTimeValueObject $since): int;
 
     public function getGivenEventsByAggregate(Uuid $aggregateId, int $offset, int $limit, string ...$events): array;
 
+    public function getEventsFilteredByAggregate(Uuid $aggregateId, int $offset, int $limit, string ...$events): array;
+
     public function countEventsForSinceVersion(Uuid $aggregateId, int $aggregateVersion): int;
 
     public function getSinceVersion(Uuid $aggregateId, int $aggregateVersion): array;

src/Repository/PostgresBaseAggregateRepository.php

@@ -169,6 +169,35 @@ protected function countByAggregateId(Uuid $aggregateId): int
         return $result['count'];
     }
 
+    protected function countGivenEventsByAggregateId(Uuid $aggregateId, string ...$eventNames): int
+    {
+        $stmt = $this->connection
+            ->createQueryBuilder()
+            ->select('count(message_id) as count')
+            ->from($this->tableName())
+            ->where('message_name IN (:eventNames)');
+
+        $stmt->setParameter('aggregateId', $aggregateId->value(), \PDO::PARAM_STR);
+        $stmt->setParameter('eventNames', $eventNames, Connection::PARAM_STR_ARRAY);
+
+        return $stmt->execute()->fetchOne();
+    }
+
+    protected function countFilteredEventsByAggregateId(Uuid $aggregateId, string ...$eventNames): int
+    {
+        $stmt = $this->connection
+            ->createQueryBuilder()
+            ->select('count(message_id) as count')
+            ->from($this->tableName())
+            ->where('message_name NOT IN (:eventNames)');
+
+        $stmt->setParameter('aggregateId', $aggregateId->value(), \PDO::PARAM_STR);
+        $stmt->setParameter('eventNames', $eventNames, Connection::PARAM_STR_ARRAY);
+
+        return $stmt->execute()->fetchOne();
+    }
+
+
     protected function countByAggregateIdSince(Uuid $aggregateId, DateTimeValueObject $since): int
     {
         $stmt = $this->connection->prepare(
@@ -264,6 +293,35 @@ protected function queryGivenEventsByAggregateIdPaginated(
         return $events;
     }
 
+    protected function queryEventsFilteredByAggregateIdPaginated(
+        Uuid $aggregateId,
+        int $offset,
+        int $limit,
+        string ...$eventNames
+    ): array {
+        $stmt = $this->connection
+            ->createQueryBuilder()
+            ->addSelect('a.message_id, a.aggregate_id, a.aggregate_version, a.occurred_on, a.message_name, a.payload')
+            ->from($this->tableName(), 'a')
+            ->where('a.aggregate_id = :aggregateId')
+            ->andWhere('a.message_name NOT IN (:eventNames)')
+            ->setParameter('aggregateId', $aggregateId->value(), \PDO::PARAM_STR)
+            ->setParameter('eventNames', $eventNames, Connection::PARAM_STR_ARRAY)
+            ->setFirstResult($offset)
+            ->setMaxResults($limit)
+            ->orderBy('a.occurred_on', 'DESC')
+            ->addOrderBy('a.aggregate_version', 'ASC')
+            ->execute();
+
+        $events = $stmt->fetchAll();
+
+        foreach ($events as $key => $event) {
+            $events[$key]['payload'] = \json_decode($event['payload'], true);
+        }
+
+        return $events;
+    }
+
     protected function execute(Statement $stmt): void
     {
         $result = $stmt->execute();

src/Repository/PostgresEventStoreRepository.php

@@ -32,6 +32,11 @@ public function getGivenEventsByAggregate(Uuid $aggregateId, int $offset, int $l
         return $this->queryGivenEventsByAggregateIdPaginated($aggregateId, $offset, $limit, ...$events);
     }
 
+    public function getEventsFilteredByAggregate(Uuid $aggregateId, int $offset, int $limit, string ...$events): array
+    {
+        return $this->queryEventsFilteredByAggregateIdPaginated($aggregateId, $offset, $limit, ...$events);
+    }
+
     public function getSince(Uuid $aggregateId, DateTimeValueObject $since): array
     {
         return $this->findByAggregateIdSince($aggregateId, $since);
@@ -62,6 +67,16 @@ public function countEventsFor(Uuid $aggregateId): int
         return $this->countByAggregateId($aggregateId);
     }
 
+    public function countGivenEventsByAggregate(Uuid $aggregateId, string ...$events): int
+    {
+        return $this->countGivenEventsByAggregateId( $aggregateId, ...$events);
+    }
+
+    public function countEventsFilteredByAggregate(Uuid $aggregateId, string ...$events): int
+    {
+        return $this->countFilteredEventsByAggregateId($aggregateId, ...$events);
+    }
+
     public function countEventsForSince(Uuid $aggregateId, DateTimeValueObject $since): int
     {
         return $this->countByAggregateIdSince($aggregateId, $since);