From 156a2c4451cc8a3475cd54c311f648ac486d7dfc Mon Sep 17 00:00:00 2001
From: Eric Landheer <ericlandheer95@gmail.com>
Date: Fri, 27 Mar 2026 08:31:45 +0100
Subject: [PATCH] Use Stream::publish() for ack-based delivery, add
 publishFireAndForget()
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Stream::put() is fire-and-forget — no delivery guarantee from JetStream.
Stream::publish() waits for an ack, ensuring the message was persisted.

publish() now uses the ack-based path as the safe default.
publishFireAndForget() retains the old put() behavior for cases where
delivery confirmation is unnecessary (logging, telemetry).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 src/Publisher/JetStreamPublisher.php       |  5 +++
 tests/Publisher/JetStreamPublisherTest.php | 43 +++++++++++++++++++++-
 2 files changed, 46 insertions(+), 2 deletions(-)

diff --git a/src/Publisher/JetStreamPublisher.php b/src/Publisher/JetStreamPublisher.php
index 6353119..aa3cc78 100644
--- a/src/Publisher/JetStreamPublisher.php
+++ b/src/Publisher/JetStreamPublisher.php
@@ -34,6 +34,11 @@ public function __construct(
     }
 
     public function publish(string $subject, string $payload): void
+    {
+        $this->stream->publish($subject, $payload);
+    }
+
+    public function publishFireAndForget(string $subject, string $payload): void
     {
         $this->stream->put($subject, $payload);
     }
diff --git a/tests/Publisher/JetStreamPublisherTest.php b/tests/Publisher/JetStreamPublisherTest.php
index 1bee387..17efc78 100644
--- a/tests/Publisher/JetStreamPublisherTest.php
+++ b/tests/Publisher/JetStreamPublisherTest.php
@@ -64,7 +64,7 @@
     new JetStreamPublisher($connection, $streamName);
 })->throws(RuntimeException::class, 'JetStream stream "MISSING_STREAM" does not exist');
 
-it('publishes using stream->put()', function () {
+it('publishes using stream->publish() for JetStream ack', function () {
     // Arrange
     $streamName = 'FLOW';
     $subject = 'flow.order.created';
@@ -91,7 +91,7 @@
 
     $stream->shouldReceive('exists')->once()->andReturn(true);
 
-    $stream->shouldReceive('put')
+    $stream->shouldReceive('publish')
         ->with($subject, $payload)
         ->once();
 
@@ -102,3 +102,42 @@
     // Assert
     expect(true)->toBeTrue();
 });
+
+it('publishes fire-and-forget using stream->put()', function () {
+    // Arrange
+    $streamName = 'FLOW';
+    $subject = 'flow.telemetry.heartbeat';
+    $payload = 'fire-and-forget-payload';
+
+    $connection = Mockery::mock(NatsConnection::class);
+    $client = Mockery::mock(Client::class);
+    $api = Mockery::mock(Api::class);
+    $stream = Mockery::mock(Stream::class);
+
+    $connection->shouldReceive('getClient')
+        ->once()
+        ->andReturn($client);
+
+    $client
+        ->shouldReceive('getApi')
+        ->once()
+        ->andReturn($api);
+
+    $api->shouldReceive('getStream')
+        ->with($streamName)
+        ->once()
+        ->andReturn($stream);
+
+    $stream->shouldReceive('exists')->once()->andReturn(true);
+
+    $stream->shouldReceive('put')
+        ->with($subject, $payload)
+        ->once();
+
+    // Act
+    $publisher = new JetStreamPublisher($connection, $streamName);
+    $publisher->publishFireAndForget($subject, $payload);
+
+    // Assert
+    expect(true)->toBeTrue();
+});