improve message serializer

src/EventBus/Serializer/DeserializeFailed.php

@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\EventBus\Serializer;
+
+use Patchlevel\EventSourcing\EventBus\Message;
+use RuntimeException;
+
+use function get_debug_type;
+use function sprintf;
+
+final class DeserializeFailed extends RuntimeException
+{
+    public static function decodeFailed(): self
+    {
+        return new self('Error while decoding message');
+    }
+
+    public static function invalidMessage(mixed $value): self
+    {
+        return new self(
+            sprintf(
+                'Value should me an instance of %s, but is %s',
+                Message::class,
+                get_debug_type($value),
+            ),
+        );
+    }
+}

src/EventBus/Serializer/PhpNativeMessageSerializer.php

@@ -8,6 +8,7 @@
 
 use function base64_decode;
 use function base64_encode;
+use function is_string;
 use function serialize;
 use function unserialize;
 
@@ -20,9 +21,21 @@ public function serialize(Message $message): string
 
     public function deserialize(string $content): Message
     {
-        return unserialize(
-            base64_decode($content),
+        $decodedString = base64_decode($content, true);
+
+        if (!is_string($decodedString)) {
+            throw DeserializeFailed::decodeFailed();
+        }
+
+        $message = unserialize(
+            $decodedString,
             ['allowed_classes' => true],
         );
+
+        if (!$message instanceof Message) {
+            throw DeserializeFailed::invalidMessage($message);
+        }
+
+        return $message;
     }
 }

src/Outbox/DoctrineOutboxStore.php

@@ -7,6 +7,7 @@
 use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\Schema\Schema;
 use Doctrine\DBAL\Types\Types;
+use Patchlevel\EventSourcing\EventBus\HeaderNotFound;
 use Patchlevel\EventSourcing\EventBus\Message;
 use Patchlevel\EventSourcing\EventBus\Serializer\MessageSerializer;
 use Patchlevel\EventSourcing\Schema\SchemaConfigurator;
@@ -71,8 +72,7 @@ public function markOutboxMessageConsumed(Message ...$messages): void
         $this->connection->transactional(
             function (Connection $connection) use ($messages): void {
                 foreach ($messages as $message) {
-                    $id = $message->customHeader(self::HEADER_OUTBOX_IDENTIFIER);
-
+                    $id = $this->extractId($message);
                     $connection->delete($this->outboxTable, ['id' => $id]);
                 }
             },
@@ -108,4 +108,19 @@ public function configureSchema(Schema $schema, Connection $connection): void
 
         $table->setPrimaryKey(['id']);
     }
+
+    private function extractId(Message $message): int
+    {
+        try {
+            $value = $message->customHeader(self::HEADER_OUTBOX_IDENTIFIER);
+        } catch (HeaderNotFound) {
+            throw OutboxHeaderIssue::missingHeader(self::HEADER_OUTBOX_IDENTIFIER);
+        }
+
+        if (!is_int($value)) {
+            throw OutboxHeaderIssue::invalidHeaderType($value);
+        }
+
+        return $value;
+    }
 }

src/Outbox/OutboxHeaderIssue.php

@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Outbox;
+
+use RuntimeException;
+
+use function gettype;
+use function sprintf;
+
+final class OutboxHeaderIssue extends RuntimeException
+{
+    public static function missingHeader(string $header): self
+    {
+        return new self(sprintf('missing header "%s"', $header));
+    }
+
+    public static function invalidHeaderType(mixed $value): self
+    {
+        return new self(sprintf('Invalid header given: need type "int" got "%s"', gettype($value)));
+    }
+}

tests/Unit/EventBus/Serializer/PhpNativeMessageSerializerTest.php

@@ -6,6 +6,7 @@
 
 use DateTimeImmutable;
 use Patchlevel\EventSourcing\EventBus\Message;
+use Patchlevel\EventSourcing\EventBus\Serializer\DeserializeFailed;
 use Patchlevel\EventSourcing\EventBus\Serializer\PhpNativeMessageSerializer;
 use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileId;
 use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileVisited;
@@ -51,6 +52,24 @@ public function testDeserialize(): void
         ], $message->headers());
     }
 
+    public function testDeserializeEncodeFailed(): void
+    {
+        $this->expectException(DeserializeFailed::class);
+
+        $nativeSerializer = new PhpNativeMessageSerializer();
+
+        $nativeSerializer->deserialize('FOO');
+    }
+
+    public function testDeserializeNotAMessage(): void
+    {
+        $this->expectException(DeserializeFailed::class);
+
+        $nativeSerializer = new PhpNativeMessageSerializer();
+
+        $nativeSerializer->deserialize('Tzo4OiJzdGRDbGFzcyI6MDp7fQ==');
+    }
+
     public function testEquals(): void
     {
         $event = new ProfileVisited(