Compatibility with updated MessageBus

src/PostgresLazyTransaction.php

@@ -0,0 +1,89 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\MessageBus\Persistence\Postgres;
+
+use Amp\Postgres\PostgresByteA;
+use Amp\Postgres\PostgresTransaction;
+use Amp\Sql\SqlTransactionError;
+use Thesis\MessageBus\Endpoint;
+use Thesis\MessageBus\Persistence\LazyTransaction;
+use Thesis\MessageBus\Persistence\OutboxAlreadyExists;
+use Thesis\MessageBus\Persistence\TransactionClosed;
+
+/**
+ * @internal
+ * @psalm-internal Thesis\MessageBus\Persistence\Postgres
+ * @implements LazyTransaction<PostgresTransaction>
+ */
+final class PostgresLazyTransaction implements LazyTransaction
+{
+    private ?PostgresTransaction $postgresTransaction = null;
+
+    public object $transaction {
+        get => $this->postgresTransaction ??= ($this->begin)();
+    }
+
+    /**
+     * @param \Closure(): PostgresTransaction $begin
+     * @param non-empty-string $outboxTable
+     */
+    public function __construct(
+        private readonly \Closure $begin,
+        private readonly string $outboxTable,
+        private readonly Endpoint $endpoint,
+    ) {}
+
+    public function recordOutboxes(array $outboxes): void
+    {
+        $placeholderGroups = [];
+        $params = [];
+
+        foreach ($outboxes as $outbox) {
+            $placeholderGroups[] = '(?, ?, ?, ?, ' . ($outbox->dispatched ? 'now()' : 'null') . ')';
+
+            $params[] = $this->endpoint->toString();
+            $params[] = $outbox->incomingMessageId;
+            $params[] = new PostgresByteA(serialize($outbox->commands));
+            $params[] = new PostgresByteA(serialize($outbox->events));
+        }
+
+        $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,
+                $params,
+            );
+        } catch (SqlTransactionError $exception) {
+            throw new TransactionClosed(previous: $exception);
+        }
+
+        if ($result->getRowCount() !== \count($outboxes)) {
+            throw new OutboxAlreadyExists();
+        }
+    }
+
+    public function commitIfBegun(): void
+    {
+        try {
+            $this->postgresTransaction?->commit();
+        } catch (SqlTransactionError $exception) {
+            throw new TransactionClosed(previous: $exception);
+        }
+    }
+
+    public function rollbackIfBegun(): void
+    {
+        try {
+            $this->postgresTransaction?->rollback();
+        } catch (SqlTransactionError $exception) {
+            throw new TransactionClosed(previous: $exception);
+        }
+    }
+}