Implement EventStoreManagement and extract interface (#11)

src/DynamoDbEventStore.php

@@ -8,11 +8,12 @@
 use AsyncAws\DynamoDb\Exception\ConditionalCheckFailedException;
 use Broadway\Domain\DomainEventStream;
 use Broadway\Domain\DomainMessage;
-use Broadway\EventStore\EventStore;
 use Broadway\EventStore\EventStreamNotFoundException;
+use Broadway\EventStore\EventVisitor;
 use Broadway\EventStore\Exception\DuplicatePlayheadException;
+use Broadway\EventStore\Management\Criteria;
 
-final class DynamoDbEventStore implements EventStore
+final class DynamoDbEventStore implements DynamoDbEventStoreInterface
 {
     public function __construct(
         private readonly DynamoDbClient $client,
@@ -93,4 +94,17 @@ public function createTable(): void
 
         $this->client->tableExists($this->inputBuilder->buildDescribeTableInput($this->table))->wait();
     }
+
+    public function visitEvents(Criteria $criteria, EventVisitor $eventVisitor): void
+    {
+        $result = $this->client->scan($this->inputBuilder->buildScanInput($this->table));
+
+        foreach ($result->getItems() as $normalizedDomainMessage) {
+            $domainMessage = $this->domainMessageNormalizer->denormalize($normalizedDomainMessage);
+
+            if ($criteria->isMatchedBy($domainMessage)) {
+                $eventVisitor->doWithEvent($domainMessage);
+            }
+        }
+    }
 }

src/DynamoDbEventStoreInterface.php

@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace chrisjenkinson\DynamoDbEventStore;
+
+use Broadway\EventStore\EventStore;
+use Broadway\EventStore\Management\EventStoreManagement;
+
+interface DynamoDbEventStoreInterface extends EventStore, EventStoreManagement
+{
+    public function createTable(): void;
+
+    public function deleteTable(): void;
+}

src/InputBuilder.php

@@ -9,6 +9,7 @@
 use AsyncAws\DynamoDb\Input\DescribeTableInput;
 use AsyncAws\DynamoDb\Input\PutItemInput;
 use AsyncAws\DynamoDb\Input\QueryInput;
+use AsyncAws\DynamoDb\Input\ScanInput;
 use AsyncAws\DynamoDb\ValueObject\AttributeDefinition;
 use AsyncAws\DynamoDb\ValueObject\AttributeValue;
 use AsyncAws\DynamoDb\ValueObject\KeySchemaElement;
@@ -44,6 +45,13 @@ public function buildQueryWithPlayheadInput(string $tableName, string $id, int $
         ]);
     }
 
+    public function buildScanInput(string $tableName): ScanInput
+    {
+        return new ScanInput([
+            'TableName' => $tableName,
+        ]);
+    }
+
     public function buildDescribeTableInput(string $tableName): DescribeTableInput
     {
         return new DescribeTableInput([

tests/DynamoDbEventStoreManagementTest.php

@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace chrisjenkinson\DynamoDbEventStore\Tests;
+
+use AsyncAws\Core\Configuration;
+use AsyncAws\DynamoDb\DynamoDbClient;
+use Broadway\EventStore\Management\EventStoreManagement;
+use Broadway\EventStore\Management\Testing\EventStoreManagementTest;
+use Broadway\Serializer\SimpleInterfaceSerializer;
+use chrisjenkinson\DynamoDbEventStore\DomainMessageNormalizer;
+use chrisjenkinson\DynamoDbEventStore\DynamoDbEventStore;
+use chrisjenkinson\DynamoDbEventStore\InputBuilder;
+use chrisjenkinson\DynamoDbEventStore\JsonDecoder;
+use chrisjenkinson\DynamoDbEventStore\JsonEncoder;
+
+final class DynamoDbEventStoreManagementTest extends EventStoreManagementTest
+{
+    protected function createEventStore(): EventStoreManagement
+    {
+        $client = new DynamoDbClient(Configuration::create([
+            'endpoint'        => 'http://dynamodb-local:8000',
+            'accessKeyId'     => '',
+            'accessKeySecret' => '',
+        ]));
+        $inputBuilder            = new InputBuilder();
+        $domainMessageNormalizer = new DomainMessageNormalizer(new SimpleInterfaceSerializer(), new SimpleInterfaceSerializer(), new JsonEncoder(), new JsonDecoder());
+
+        $eventStore = new DynamoDbEventStore($client, $inputBuilder, $domainMessageNormalizer, 'table');
+
+        $eventStore->deleteTable();
+        $eventStore->createTable();
+
+        return $eventStore;
+    }
+}