Add recorded_at, dispatched_at columns

Author: vudaltsov

composer.lock

@@ -33,9 +33,10 @@ public function setup(): void
                 (
                     endpoint text not null,
                     incoming_message_id text not null,
+                    recorded_at timestamptz not null default now(),
                     commands bytea not null,
                     events bytea not null,
-                    dispatched bool not null default false,
+                    dispatched_at timestamptz default null,
                     primary key (incoming_message_id, endpoint)
                 )
                 SQL,
@@ -59,7 +60,7 @@ public function findOutbox(string $endpoint, string $incomingMessageId): ?Outbox
             ->postgres
             ->execute(
                 <<<SQL
-                    select commands, events, dispatched
+                    select commands, events, dispatched_at is not null as dispatched 
                     from {$this->outboxTable}
                     where endpoint = ? and incoming_message_id = ?
                     SQL,
@@ -88,8 +89,8 @@ public function markOutboxDispatched(string $endpoint, string $incomingMessageId
         $this->postgres->execute(
             <<<SQL
                 update {$this->outboxTable}
-                set dispatched = true
-                where endpoint = ? and incoming_message_id = ? and not dispatched
+                set dispatched_at = now()
+                where endpoint = ? and incoming_message_id = ? and dispatched_at is null
                 SQL,
             [
                 $endpoint,

src/PostgresStorage.php

@@ -30,21 +30,22 @@ public function __construct(
         private string $incomingMessageId,
     ) {}
 
-    public function insertOutbox(Outbox $outbox): void
+    public function recordOutbox(Outbox $outbox): void
     {
+        $dispatched = $outbox->dispatched ? 'now()' : 'null';
+
         try {
             $result = $this->wrappedTransaction->execute(
                 <<<SQL
-                    insert into {$this->outboxTable} (endpoint, incoming_message_id, commands, events, dispatched)
-                    values (?, ?, ?, ?, ?)
+                    insert into {$this->outboxTable} (endpoint, incoming_message_id, commands, events, dispatched_at)
+                    values (?, ?, ?, ?, {$dispatched})
                     on conflict (endpoint, incoming_message_id) do nothing
                     SQL,
                 [
                     $this->endpoint,
                     $this->incomingMessageId,
                     new PostgresByteA(serialize($outbox->commands)),
                     new PostgresByteA(serialize($outbox->events)),
-                    $outbox->dispatched,
                 ],
             );
         } catch (SqlTransactionError $exception) {