From 936d731146605d00251a96b0e0d9ff157a6d4c6b Mon Sep 17 00:00:00 2001 From: Willem-Jan Zijderveld Date: Fri, 6 Sep 2024 20:32:08 +0200 Subject: [PATCH 1/2] Use correct ParameterTypes when dealing with binary uuids --- .../DefaultDoctrineMessageRepositoryTest.php | 1 + .../DoctrineMessageRepository.php | 25 +++++++++++++++---- .../DoctrineMessageRepositoryTestCase.php | 22 ++++++++++++++++ .../DoctrineUuidV4MessageRepository.php | 19 ++++++++++---- .../DefaultDoctrineMessageRepositoryTest.php | 1 - .../DoctrineMessageRepository.php | 17 ++++++++++--- .../DoctrineMessageRepositoryTestCase.php | 22 ++++++++++++++++ 7 files changed, 93 insertions(+), 14 deletions(-) diff --git a/src/DoctrineMessageRepository/DefaultDoctrineMessageRepositoryTest.php b/src/DoctrineMessageRepository/DefaultDoctrineMessageRepositoryTest.php index fff2e16..e4c6480 100644 --- a/src/DoctrineMessageRepository/DefaultDoctrineMessageRepositoryTest.php +++ b/src/DoctrineMessageRepository/DefaultDoctrineMessageRepositoryTest.php @@ -18,6 +18,7 @@ protected function messageRepository(): DoctrineMessageRepository serializer: new ConstructingMessageSerializer(), tableSchema: new DefaultTableSchema(), aggregateRootIdEncoder: new BinaryUuidIdEncoder(), + eventIdEncoder: new BinaryUuidIdEncoder(), ); } } diff --git a/src/DoctrineMessageRepository/DoctrineMessageRepository.php b/src/DoctrineMessageRepository/DoctrineMessageRepository.php index 8b8fcca..bd7ce79 100644 --- a/src/DoctrineMessageRepository/DoctrineMessageRepository.php +++ b/src/DoctrineMessageRepository/DoctrineMessageRepository.php @@ -3,6 +3,7 @@ namespace EventSauce\MessageRepository\DoctrineMessageRepository; use Doctrine\DBAL\Connection; +use Doctrine\DBAL\ParameterType; use Doctrine\DBAL\Query\QueryBuilder; use EventSauce\IdEncoding\BinaryUuidIdEncoder; use EventSauce\IdEncoding\IdEncoder; @@ -73,9 +74,11 @@ public function persist(Message ...$messages): void $payload = $this->serializer->serializeMessage($message); $payload['headers'][Header::EVENT_ID] ??= Uuid::uuid4()->toString(); + $eventIdIndex = $this->indexParameter('event_id', $index); + $aggregateRootIdIndex = $this->indexParameter('aggregate_root_id', $index); $messageParameters = [ - $this->indexParameter('event_id', $index) => $this->eventIdEncoder->encodeId($payload['headers'][Header::EVENT_ID]), - $this->indexParameter('aggregate_root_id', $index) => $this->aggregateRootIdEncoder->encodeId($message->aggregateRootId()), + $eventIdIndex => $this->eventIdEncoder->encodeId($payload['headers'][Header::EVENT_ID]), + $aggregateRootIdIndex => $this->aggregateRootIdEncoder->encodeId($message->aggregateRootId()), $this->indexParameter('version', $index) => $payload['headers'][Header::AGGREGATE_ROOT_VERSION] ?? 0, $this->indexParameter('payload', $index) => json_encode($payload, $this->jsonEncodeOptions), ]; @@ -98,8 +101,16 @@ public function persist(Message ...$messages): void implode("),\n(", $insertValues), ); + $types = []; + if ($this->eventIdEncoder instanceof BinaryUuidIdEncoder) { + $types[$eventIdIndex] = ParameterType::BINARY; + } + if ($this->aggregateRootIdEncoder instanceof BinaryUuidIdEncoder) { + $types[$aggregateRootIdIndex] = ParameterType::BINARY; + } + try { - $this->connection->executeStatement($insertQuery, $insertParameters); + $this->connection->executeStatement($insertQuery, $insertParameters, $types); } catch (Throwable $exception) { throw UnableToPersistMessages::dueTo('', $exception); } @@ -117,9 +128,11 @@ private function formatNamedParameters(array $parameters): array public function retrieveAll(AggregateRootId $id): Generator { + $aggregateRootIdType = $this->aggregateRootIdEncoder instanceof BinaryUuidIdEncoder ? ParameterType::BINARY : ParameterType::STRING; + $builder = $this->createQueryBuilder(); $builder->where(sprintf('%s = :aggregate_root_id', $this->tableSchema->aggregateRootIdColumn())); - $builder->setParameter('aggregate_root_id', $this->aggregateRootIdEncoder->encodeId($id)); + $builder->setParameter('aggregate_root_id', $this->aggregateRootIdEncoder->encodeId($id), $aggregateRootIdType); try { return $this->yieldMessagesFromPayloads($builder->executeQuery()->iterateColumn()); @@ -133,10 +146,12 @@ public function retrieveAll(AggregateRootId $id): Generator */ public function retrieveAllAfterVersion(AggregateRootId $id, int $aggregateRootVersion): Generator { + $aggregateRootIdType = $this->aggregateRootIdEncoder instanceof BinaryUuidIdEncoder ? ParameterType::BINARY : ParameterType::STRING; + $builder = $this->createQueryBuilder(); $builder->where(sprintf('%s = :aggregate_root_id', $this->tableSchema->aggregateRootIdColumn())); $builder->andWhere(sprintf('%s > :version', $this->tableSchema->versionColumn())); - $builder->setParameter('aggregate_root_id', $this->aggregateRootIdEncoder->encodeId($id)); + $builder->setParameter('aggregate_root_id', $this->aggregateRootIdEncoder->encodeId($id), $aggregateRootIdType); $builder->setParameter('version', $aggregateRootVersion); try { diff --git a/src/DoctrineMessageRepository/DoctrineMessageRepositoryTestCase.php b/src/DoctrineMessageRepository/DoctrineMessageRepositoryTestCase.php index 06fe22e..b233411 100644 --- a/src/DoctrineMessageRepository/DoctrineMessageRepositoryTestCase.php +++ b/src/DoctrineMessageRepository/DoctrineMessageRepositoryTestCase.php @@ -32,6 +32,28 @@ protected function setUp(): void $this->truncateTable(); } + protected function tearDown(): void + { + parent::tearDown(); + if (str_contains($this->formatDsn(), 'pgsql')) { + // can only check this for mysql + return; + } + + $warning = $this->connection->executeQuery('SHOW WARNINGS')->fetchNumeric(); + if ($warning !== false && count($warning) > 0) { + if (str_contains($warning[2], 'Base table or view not found') || str_contains($warning[2], "doesn't exist")) { + // shortcut for tests + return; + } + self::fail(sprintf( + 'Warnings issued durings tests, these can potentially result in data loss: [%d] %s', + $warning[1], + $warning[2], + )); + } + } + protected function formatDsn(): string { $host = getenv('EVENTSAUCE_TESTING_MYSQL_HOST') ?: '127.0.0.1'; diff --git a/src/DoctrineMessageRepository/DoctrineUuidV4MessageRepository.php b/src/DoctrineMessageRepository/DoctrineUuidV4MessageRepository.php index e9311b7..029ff3e 100644 --- a/src/DoctrineMessageRepository/DoctrineUuidV4MessageRepository.php +++ b/src/DoctrineMessageRepository/DoctrineUuidV4MessageRepository.php @@ -3,6 +3,7 @@ namespace EventSauce\MessageRepository\DoctrineMessageRepository; use Doctrine\DBAL\Connection; +use Doctrine\DBAL\ParameterType; use Doctrine\DBAL\Query\QueryBuilder; use EventSauce\EventSourcing\AggregateRootId; use EventSauce\EventSourcing\Header; @@ -74,9 +75,11 @@ public function persist(Message ...$messages): void $payload = $this->serializer->serializeMessage($message); $payload['headers'][Header::EVENT_ID] ??= Uuid::uuid4()->toString(); + $eventIdIndex = $this->indexParameter('event_id', $index); + $aggregateRootIdIndex = $this->indexParameter('aggregate_root_id', $index); $messageParameters = [ - $this->indexParameter('event_id', $index) => $this->uuidEncoder->encodeString($payload['headers'][Header::EVENT_ID]), - $this->indexParameter('aggregate_root_id', $index) => $this->uuidEncoder->encodeString($payload['headers'][Header::AGGREGATE_ROOT_ID]), + $eventIdIndex => $this->uuidEncoder->encodeString($payload['headers'][Header::EVENT_ID]), + $aggregateRootIdIndex => $this->uuidEncoder->encodeString($payload['headers'][Header::AGGREGATE_ROOT_ID]), $this->indexParameter('version', $index) => $payload['headers'][Header::AGGREGATE_ROOT_VERSION] ?? 0, $this->indexParameter('payload', $index) => json_encode($payload, $this->jsonEncodeOptions), ]; @@ -99,8 +102,14 @@ public function persist(Message ...$messages): void implode("),\n(", $insertValues), ); + $types = []; + if ($this->uuidEncoder instanceof BinaryUuidEncoder) { + $types[$eventIdIndex] = ParameterType::BINARY; + $types[$aggregateRootIdIndex] = ParameterType::BINARY; + } + try { - $this->connection->executeStatement($insertQuery, $insertParameters); + $this->connection->executeStatement($insertQuery, $insertParameters, $types); } catch (Throwable $exception) { throw UnableToPersistMessages::dueTo('', $exception); } @@ -120,7 +129,7 @@ public function retrieveAll(AggregateRootId $id): Generator { $builder = $this->createQueryBuilder(); $builder->where(sprintf('%s = :aggregate_root_id', $this->tableSchema->aggregateRootIdColumn())); - $builder->setParameter('aggregate_root_id', $this->uuidEncoder->encodeString($id->toString())); + $builder->setParameter('aggregate_root_id', $this->uuidEncoder->encodeString($id->toString()), $this->uuidEncoder instanceof BinaryUuidEncoder ? ParameterType::BINARY : ParameterType::STRING); try { return $this->yieldMessagesFromPayloads($builder->executeQuery()->iterateColumn()); @@ -137,7 +146,7 @@ public function retrieveAllAfterVersion(AggregateRootId $id, int $aggregateRootV $builder = $this->createQueryBuilder(); $builder->where(sprintf('%s = :aggregate_root_id', $this->tableSchema->aggregateRootIdColumn())); $builder->andWhere(sprintf('%s > :version', $this->tableSchema->versionColumn())); - $builder->setParameter('aggregate_root_id', $this->uuidEncoder->encodeString($id->toString())); + $builder->setParameter('aggregate_root_id', $this->uuidEncoder->encodeString($id->toString()), $this->uuidEncoder instanceof BinaryUuidEncoder ? ParameterType::BINARY : ParameterType::STRING); $builder->setParameter('version', $aggregateRootVersion); try { diff --git a/src/DoctrineV2MessageRepository/DefaultDoctrineMessageRepositoryTest.php b/src/DoctrineV2MessageRepository/DefaultDoctrineMessageRepositoryTest.php index d14dcf9..f19641b 100644 --- a/src/DoctrineV2MessageRepository/DefaultDoctrineMessageRepositoryTest.php +++ b/src/DoctrineV2MessageRepository/DefaultDoctrineMessageRepositoryTest.php @@ -6,7 +6,6 @@ use EventSauce\EventSourcing\Serialization\ConstructingMessageSerializer; use EventSauce\MessageRepository\DoctrineV2MessageRepository\DoctrineMessageRepositoryTestCase; use EventSauce\MessageRepository\TableSchema\DefaultTableSchema; -use EventSauce\UuidEncoding\BinaryUuidEncoder; /** * @group doctrine2 diff --git a/src/DoctrineV2MessageRepository/DoctrineMessageRepository.php b/src/DoctrineV2MessageRepository/DoctrineMessageRepository.php index d1ecff2..063df43 100644 --- a/src/DoctrineV2MessageRepository/DoctrineMessageRepository.php +++ b/src/DoctrineV2MessageRepository/DoctrineMessageRepository.php @@ -5,6 +5,7 @@ use Doctrine\DBAL\Connection; use Doctrine\DBAL\Driver\ResultStatement; use Doctrine\DBAL\ForwardCompatibility\Result; +use Doctrine\DBAL\ParameterType; use Doctrine\DBAL\Query\QueryBuilder; use EventSauce\EventSourcing\AggregateRootId; use EventSauce\EventSourcing\Header; @@ -75,9 +76,11 @@ public function persist(Message ...$messages): void $payload = $this->serializer->serializeMessage($message); $payload['headers'][Header::EVENT_ID] ??= Uuid::uuid4()->toString(); + $eventIdIndex = $this->indexParameter('event_id', $index); + $aggregateRootIdIndex = $this->indexParameter('aggregate_root_id', $index); $messageParameters = [ - $this->indexParameter('event_id', $index) => $this->eventIdEncoder->encodeId($payload['headers'][Header::EVENT_ID]), - $this->indexParameter('aggregate_root_id', $index) => $this->aggregateRootIdEncoder->encodeId($message->aggregateRootId()), + $eventIdIndex => $this->eventIdEncoder->encodeId($payload['headers'][Header::EVENT_ID]), + $aggregateRootIdIndex => $this->aggregateRootIdEncoder->encodeId($message->aggregateRootId()), $this->indexParameter('version', $index) => $payload['headers'][Header::AGGREGATE_ROOT_VERSION] ?? 0, $this->indexParameter('payload', $index) => json_encode($payload, $this->jsonEncodeOptions), ]; @@ -100,8 +103,16 @@ public function persist(Message ...$messages): void implode("),\n(", $insertValues), ); + $types = []; + if ($this->eventIdEncoder instanceof BinaryUuidIdEncoder) { + $types[$eventIdIndex] = ParameterType::BINARY; + } + if ($this->aggregateRootIdEncoder instanceof BinaryUuidIdEncoder) { + $types[$aggregateRootIdIndex] = ParameterType::BINARY; + } + try { - $this->connection->executeStatement($insertQuery, $insertParameters); + $this->connection->executeStatement($insertQuery, $insertParameters, $types); } catch (Throwable $exception) { throw UnableToPersistMessages::dueTo('', $exception); } diff --git a/src/DoctrineV2MessageRepository/DoctrineMessageRepositoryTestCase.php b/src/DoctrineV2MessageRepository/DoctrineMessageRepositoryTestCase.php index ceb42f3..35d26b8 100644 --- a/src/DoctrineV2MessageRepository/DoctrineMessageRepositoryTestCase.php +++ b/src/DoctrineV2MessageRepository/DoctrineMessageRepositoryTestCase.php @@ -30,6 +30,28 @@ protected function setUp(): void $this->connection->executeQuery('TRUNCATE TABLE ' . $this->tableName); } + protected function tearDown(): void + { + parent::tearDown(); + if (str_contains($this->formatDsn(), 'pgsql')) { + // can only check this for mysql + return; + } + + $warning = $this->connection->executeQuery('SHOW WARNINGS')->fetchNumeric(); + if (count($warning) > 0) { + if (str_contains($warning[2], 'Base table or view not found') || str_contains($warning[2], "doesn't exist")) { + // shortcut for tests + return; + } + self::fail(sprintf( + 'Warnings issued durings tests, these can potentially result in data loss: [%d] %s', + $warning[1], + $warning[2], + )); + } + } + protected function aggregateRootId(): AggregateRootId { return DummyAggregateRootId::generate(); From aa11562b77b48c4717022a8d412df18202385458 Mon Sep 17 00:00:00 2001 From: Willem-Jan Zijderveld Date: Tue, 28 Oct 2025 16:05:10 +0100 Subject: [PATCH 2/2] Backport fix to DoctrineV2 --- .../DoctrineMessageRepository.php | 4 ++-- .../DoctrineMessageRepositoryTestCase.php | 2 +- .../DoctrineUuidV4MessageRepository.php | 19 ++++++++++++++----- 3 files changed, 17 insertions(+), 8 deletions(-) diff --git a/src/DoctrineV2MessageRepository/DoctrineMessageRepository.php b/src/DoctrineV2MessageRepository/DoctrineMessageRepository.php index 063df43..e1833c7 100644 --- a/src/DoctrineV2MessageRepository/DoctrineMessageRepository.php +++ b/src/DoctrineV2MessageRepository/DoctrineMessageRepository.php @@ -132,7 +132,7 @@ public function retrieveAll(AggregateRootId $id): Generator { $builder = $this->createQueryBuilder(); $builder->where(sprintf('%s = :aggregate_root_id', $this->tableSchema->aggregateRootIdColumn())); - $builder->setParameter('aggregate_root_id', $this->aggregateRootIdEncoder->encodeId($id)); + $builder->setParameter('aggregate_root_id', $this->aggregateRootIdEncoder->encodeId($id), $this->aggregateRootIdEncoder instanceof BinaryUuidIdEncoder ? ParameterType::BINARY : ParameterType::STRING); try { /** @var ResultStatement $resultStatement */ @@ -152,7 +152,7 @@ public function retrieveAllAfterVersion(AggregateRootId $id, int $aggregateRootV $builder = $this->createQueryBuilder(); $builder->where(sprintf('%s = :aggregate_root_id', $this->tableSchema->aggregateRootIdColumn())); $builder->andWhere(sprintf('%s > :version', $this->tableSchema->versionColumn())); - $builder->setParameter('aggregate_root_id', $this->aggregateRootIdEncoder->encodeId($id)); + $builder->setParameter('aggregate_root_id', $this->aggregateRootIdEncoder->encodeId($id), $this->aggregateRootIdEncoder instanceof BinaryUuidIdEncoder ? ParameterType::BINARY : ParameterType::STRING); $builder->setParameter('version', $aggregateRootVersion); try { diff --git a/src/DoctrineV2MessageRepository/DoctrineMessageRepositoryTestCase.php b/src/DoctrineV2MessageRepository/DoctrineMessageRepositoryTestCase.php index 35d26b8..7248142 100644 --- a/src/DoctrineV2MessageRepository/DoctrineMessageRepositoryTestCase.php +++ b/src/DoctrineV2MessageRepository/DoctrineMessageRepositoryTestCase.php @@ -39,7 +39,7 @@ protected function tearDown(): void } $warning = $this->connection->executeQuery('SHOW WARNINGS')->fetchNumeric(); - if (count($warning) > 0) { + if ($warning !== false && count($warning) > 0) { if (str_contains($warning[2], 'Base table or view not found') || str_contains($warning[2], "doesn't exist")) { // shortcut for tests return; diff --git a/src/DoctrineV2MessageRepository/DoctrineUuidV4MessageRepository.php b/src/DoctrineV2MessageRepository/DoctrineUuidV4MessageRepository.php index 7b14491..21ed365 100644 --- a/src/DoctrineV2MessageRepository/DoctrineUuidV4MessageRepository.php +++ b/src/DoctrineV2MessageRepository/DoctrineUuidV4MessageRepository.php @@ -5,6 +5,7 @@ use Doctrine\DBAL\Connection; use Doctrine\DBAL\Driver\ResultStatement; use Doctrine\DBAL\ForwardCompatibility\Result; +use Doctrine\DBAL\ParameterType; use Doctrine\DBAL\Query\QueryBuilder; use EventSauce\EventSourcing\AggregateRootId; use EventSauce\EventSourcing\Header; @@ -78,9 +79,11 @@ public function persist(Message ...$messages): void $payload = $this->serializer->serializeMessage($message); $payload['headers'][Header::EVENT_ID] ??= Uuid::uuid4()->toString(); + $eventIdIndex = $this->indexParameter('event_id', $index); + $aggregateRootIdIndex = $this->indexParameter('aggregate_root_id', $index); $messageParameters = [ - $this->indexParameter('event_id', $index) => $this->uuidEncoder->encodeString($payload['headers'][Header::EVENT_ID]), - $this->indexParameter('aggregate_root_id', $index) => $this->uuidEncoder->encodeString($payload['headers'][Header::AGGREGATE_ROOT_ID]), + $eventIdIndex => $this->uuidEncoder->encodeString($payload['headers'][Header::EVENT_ID]), + $aggregateRootIdIndex => $this->uuidEncoder->encodeString($payload['headers'][Header::AGGREGATE_ROOT_ID]), $this->indexParameter('version', $index) => $payload['headers'][Header::AGGREGATE_ROOT_VERSION] ?? 0, $this->indexParameter('payload', $index) => json_encode($payload, $this->jsonEncodeOptions), ]; @@ -103,8 +106,14 @@ public function persist(Message ...$messages): void implode("),\n(", $insertValues), ); + $types = []; + if ($this->uuidEncoder instanceof BinaryUuidEncoder) { + $types[$eventIdIndex] = ParameterType::BINARY; + $types[$aggregateRootIdIndex] = ParameterType::BINARY; + } + try { - $this->connection->executeStatement($insertQuery, $insertParameters); + $this->connection->executeStatement($insertQuery, $insertParameters, $types); } catch (Throwable $exception) { throw UnableToPersistMessages::dueTo('', $exception); } @@ -124,7 +133,7 @@ public function retrieveAll(AggregateRootId $id): Generator { $builder = $this->createQueryBuilder(); $builder->where(sprintf('%s = :aggregate_root_id', $this->tableSchema->aggregateRootIdColumn())); - $builder->setParameter('aggregate_root_id', $this->uuidEncoder->encodeString($id->toString())); + $builder->setParameter('aggregate_root_id', $this->uuidEncoder->encodeString($id->toString()), $this->uuidEncoder instanceof BinaryUuidEncoder ? ParameterType::BINARY : ParameterType::STRING); try { /** @var ResultStatement $resultStatement */ @@ -144,7 +153,7 @@ public function retrieveAllAfterVersion(AggregateRootId $id, int $aggregateRootV $builder = $this->createQueryBuilder(); $builder->where(sprintf('%s = :aggregate_root_id', $this->tableSchema->aggregateRootIdColumn())); $builder->andWhere(sprintf('%s > :version', $this->tableSchema->versionColumn())); - $builder->setParameter('aggregate_root_id', $this->uuidEncoder->encodeString($id->toString())); + $builder->setParameter('aggregate_root_id', $this->uuidEncoder->encodeString($id->toString()), $this->uuidEncoder instanceof BinaryUuidEncoder ? ParameterType::BINARY : ParameterType::STRING); $builder->setParameter('version', $aggregateRootVersion); try {