Imrpovements after review

Author: vudaltsov

src/PostgresLazyTransaction.php

@@ -8,9 +8,8 @@
 use Amp\Postgres\PostgresTransaction;
 use Amp\Sql\SqlTransactionError;
 use Thesis\MessageBus\Endpoint;
-use Thesis\MessageBus\Persistence\Outbox;
-use Thesis\MessageBus\Persistence\OutboxAlreadyExists;
 use Thesis\MessageBus\Persistence\LazyTransaction;
+use Thesis\MessageBus\Persistence\OutboxAlreadyExists;
 use Thesis\MessageBus\Persistence\TransactionClosed;
 
 /**
@@ -31,42 +30,41 @@ final class PostgresLazyTransaction implements LazyTransaction
      * @param non-empty-string $outboxTable
      */
     public function __construct(
-        private \Closure $begin,
-        private string $outboxTable,
-        private Endpoint $endpoint,
+        private readonly \Closure $begin,
+        private readonly string $outboxTable,
+        private readonly Endpoint $endpoint,
     ) {}
 
     public function recordOutboxes(array $outboxes): void
     {
-        $insertData = [];
-
-        $placeholders = [];
+        $placeholderGroups = [];
+        $params = [];
 
         foreach ($outboxes as $outbox) {
-            $insertData[] = $this->endpoint->toString();
-            $insertData[] = $outbox->incomingMessageId;
-            $insertData[] = new PostgresByteA(serialize($outbox->commands));
-            $insertData[] = new PostgresByteA(serialize($outbox->events));
+            $placeholderGroups[] = '(?, ?, ?, ?, ' . ($outbox->dispatched ? 'now()' : 'null') . ')';
 
-            $placeholders[] = '(?, ?, ?, ?,' . ($outbox->dispatched ? 'now()' : 'null') . ')';
+            $params[] = $this->endpoint->toString();
+            $params[] = $outbox->incomingMessageId;
+            $params[] = new PostgresByteA(serialize($outbox->commands));
+            $params[] = new PostgresByteA(serialize($outbox->events));
         }
 
-        try {
-            $placeholders = implode(',', $placeholders);
+        $placeholders = implode(',', $placeholderGroups);
 
+        try {
             $result = $this->transaction->execute(
                 <<<SQL
                     insert into {$this->outboxTable} (endpoint, incoming_message_id, commands, events, dispatched_at)
                     values {$placeholders}
                     on conflict (endpoint, incoming_message_id) do nothing
                     SQL,
-                $insertData,
+                $params,
             );
         } catch (SqlTransactionError $exception) {
             throw new TransactionClosed(previous: $exception);
         }
 
-        if ($result->getRowCount() !== count($outboxes)) {
+        if ($result->getRowCount() !== \count($outboxes)) {
             throw new OutboxAlreadyExists();
         }
     }

src/PostgresStorage.php

@@ -5,23 +5,21 @@
 namespace Thesis\MessageBus\Persistence\Postgres;
 
 use Amp\Postgres\PostgresLink;
-use Amp\Postgres\PostgresTransaction as AmphpPostgresTransaction;
+use Amp\Postgres\PostgresTransaction;
 use Thesis\MessageBus\Endpoint;
 use Thesis\MessageBus\Envelope;
+use Thesis\MessageBus\Persistence\LazyTransaction;
 use Thesis\MessageBus\Persistence\Outbox;
 use Thesis\MessageBus\Persistence\Storage;
-use Thesis\MessageBus\Persistence\LazyTransaction;
 
 /**
  * @api
- * @implements Storage<AmphpPostgresTransaction>
+ * @implements Storage<PostgresTransaction>
  */
 final class PostgresStorage implements Storage
 {
     public array $transactionClasses {
-        get {
-            return [AmphpPostgresTransaction::class];
-        }
+        get => [PostgresTransaction::class];
     }
 
     /**
@@ -61,26 +59,30 @@ public function createLazyTransaction(Endpoint $endpoint): LazyTransaction
 
     public function findOutboxes(Endpoint $endpoint, array $incomingMessageIds): array
     {
-        $placeholders = implode(',', array_fill(0, count($incomingMessageIds), '?'));
+        $placeholders = str_repeat('?, ', \count($incomingMessageIds) - 1) . '?';
 
         $result = $this
             ->postgres
             ->execute(
                 <<<SQL
-                    select incoming_message_id, commands, events, dispatched_at is not null as dispatched 
+                    select
+                        incoming_message_id,
+                        commands,
+                        events,
+                        dispatched_at is not null as dispatched 
                     from {$this->outboxTable}
                     where endpoint = ? and incoming_message_id in ({$placeholders})
                     SQL,
-                [$endpoint->toString(), ...$incomingMessageIds],
+                [
+                    $endpoint->toString(),
+                    ...$incomingMessageIds,
+                ],
             );
 
         $outboxes = [];
 
         while (null !== $row = $result->fetchRow()) {
-            /**
-             * @var array{commands: string, events: string, incoming_message_id: non-empty-string, dispatched: bool} $row
-             */
-
+            /** @var array{commands: string, events: string, incoming_message_id: non-empty-string, dispatched: bool} $row */
             /** @var list<Envelope> $commands */
             $commands = unserialize($row['commands']);
             /** @var list<Envelope> $events */
@@ -99,13 +101,15 @@ public function findOutboxes(Endpoint $endpoint, array $incomingMessageIds): arr
 
     public function markOutboxesDispatched(Endpoint $endpoint, array $incomingMessageIds): void
     {
-        $placeholders = implode(',', array_fill(0, count($incomingMessageIds), '?'));
+        $placeholders = str_repeat('?, ', \count($incomingMessageIds) - 1) . '?';
 
         $this->postgres->execute(
             <<<SQL
                 update {$this->outboxTable}
                 set dispatched_at = now()
-                where endpoint = ? and incoming_message_id = ({$placeholders}) and dispatched_at is null
+                where endpoint = ?
+                  and incoming_message_id in ({$placeholders})
+                  and dispatched_at is null
                 SQL,
             [
                 $endpoint->toString(),