Pod-Point · clemblanco · Oct 9, 2023 · Sep 28, 2023 · Oct 6, 2023 · Oct 9, 2023
diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml
@@ -4,7 +4,7 @@ on: [push, pull_request]
 
 jobs:
   test:
-    runs-on: ubuntu-18.04
+    runs-on: ubuntu-latest
     strategy:
       fail-fast: true
       matrix:

diff --git a/src/Pub/Broadcasting/Broadcasters/EventBridgeBroadcaster.php b/src/Pub/Broadcasting/Broadcasters/EventBridgeBroadcaster.php
@@ -4,6 +4,8 @@
 
 use Aws\EventBridge\EventBridgeClient;
 use Illuminate\Broadcasting\Broadcasters\Broadcaster;
+use Illuminate\Support\Arr;
+use Illuminate\Support\Facades\Log;
 
 class EventBridgeBroadcaster extends Broadcaster
 {
@@ -52,9 +54,17 @@ public function broadcast(array $channels, $event, array $payload = [])
     {
         $events = $this->mapToEventBridgeEntries($channels, $event, $payload);
 
-        $this->eventBridgeClient->putEvents([
+        $result = $this->eventBridgeClient->putEvents([
             'Entries' => $events,
         ]);
+
+        if ($this->failedToBroadcast($result)) {
+            Log::error('Failed to send events to EventBridge', [
+                'errors' => collect($result->get('Entries'))->filter(function (array $entry) {
+                    return Arr::has($entry, 'ErrorCode') || Arr::has($entry, 'ErrorMessage');
+                })->toArray(),
+            ]);
+        }
     }
 
     /**
@@ -76,4 +86,11 @@ protected function mapToEventBridgeEntries(array $channels, string $event, array
             })
             ->all();
     }
+
+    protected function failedToBroadcast(?\Aws\Result $result): bool
+    {
+        return $result
+            && $result->hasKey('FailedEntryCount')
+            && $result->get('FailedEntryCount') > 0;
+    }
 }
diff --git a/src/Sub/Queue/Jobs/SnsEventDispatcherJob.php b/src/Sub/Queue/Jobs/SnsEventDispatcherJob.php
@@ -21,6 +21,8 @@ public function fire()
                     'message delivery is disabled for your SQS subscription.', $this->job);
             }
 
+            $this->release();
+
             return;
         }
 
@@ -30,6 +32,8 @@ public function fire()
                 'subject' => $this->snsSubject(),
             ]);
         }
+
+        $this->delete();
     }
 
     /**

diff --git a/tests/EventServiceProviderTest.php b/tests/EventServiceProviderTest.php
@@ -116,6 +116,7 @@ public function invalidCredentialsDataProvider()
 
     /**
      * @test
+     *
      * @dataProvider invalidCredentialsDataProvider
      */
     public function it_can_make_sure_some_aws_credentials_are_provided_and_valid(array $invalidCredentials)

diff --git a/tests/Pub/BasicEvents/EventBridgeTest.php b/tests/Pub/BasicEvents/EventBridgeTest.php
@@ -2,8 +2,9 @@
 
 namespace PodPoint\AwsPubSub\Tests\Pub\BasicEvents;
 
-use Illuminate\Foundation\Application;
-use Mockery;
+use Aws\Result;
+use Illuminate\Support\Facades\Log;
+use Mockery as m;
 use Mockery\MockInterface;
 use PodPoint\AwsPubSub\Tests\Pub\Concerns\InteractsWithEventBridge;
 use PodPoint\AwsPubSub\Tests\Pub\TestClasses\Events\UserRetrieved;
@@ -30,60 +31,52 @@ protected function getEnvironmentSetUp($app)
     /** @test */
     public function it_broadcasts_basic_event_with_the_event_name_as_the_detail_type_and_serialised_event_as_the_detail()
     {
-        $jane = $this->createJane();
+        $event = new UserRetrieved($this->createJane());
 
-        $janeRetrieved = new UserRetrieved($jane);
-
-        $this->mockEventBridge(function (MockInterface $eventBridge) use ($janeRetrieved) {
+        $this->mockEventBridge(function (MockInterface $eventBridge) use ($event) {
             $eventBridge
                 ->shouldReceive('putEvents')
                 ->once()
-                ->with(Mockery::on(function ($arg) use ($janeRetrieved) {
-                    return $arg['Entries'][0]['Detail'] === json_encode($janeRetrieved)
+                ->with(m::on(function ($arg) use ($event) {
+                    return $arg['Entries'][0]['Detail'] === json_encode($event)
                         && $arg['Entries'][0]['DetailType'] === UserRetrieved::class
                         && $arg['Entries'][0]['EventBusName'] === 'users'
                         && $arg['Entries'][0]['Source'] === 'my-app';
                 }));
         });
 
-        event($janeRetrieved);
+        event($event);
     }
 
     /** @test */
     public function it_broadcasts_basic_event_with_action()
     {
-        $jane = $this->createJane();
-
-        $janeRetrieved = new UserRetrievedWithCustomName($jane);
+        $event = new UserRetrievedWithCustomName($this->createJane());
 
-        $this->mockEventBridge(function (MockInterface $eventBridge) use ($janeRetrieved) {
+        $this->mockEventBridge(function (MockInterface $eventBridge) use ($event) {
             $eventBridge
                 ->shouldReceive('putEvents')
                 ->once()
-                ->with(Mockery::on(function ($arg) use ($janeRetrieved) {
-                    return $arg['Entries'][0]['Detail'] === json_encode($janeRetrieved)
-                        && $arg['Entries'][0]['DetailType'] === 'user.retrieved'
-                        && $arg['Entries'][0]['EventBusName'] === 'users'
-                        && $arg['Entries'][0]['Source'] === 'my-app';
+                ->with(m::on(function ($arg) use ($event) {
+                    return $arg['Entries'][0]['Detail'] === json_encode($event)
+                        && $arg['Entries'][0]['DetailType'] === 'user.retrieved';
                 }));
         });
 
-        event($janeRetrieved);
+        event($event);
     }
 
     /** @test */
     public function it_broadcasts_basic_event_with_action_and_custom_payload()
     {
-        $jane = $this->createJane();
+        $event = new UserRetrievedWithCustomPayload($this->createJane());
 
-        $janeRetrieved = new UserRetrievedWithCustomPayload($jane);
-
-        $this->mockEventBridge(function (MockInterface $eventBridge) use ($janeRetrieved) {
+        $this->mockEventBridge(function (MockInterface $eventBridge) use ($event) {
             $eventBridge
                 ->shouldReceive('putEvents')
                 ->once()
-                ->with(Mockery::on(function ($arg) use ($janeRetrieved) {
-                    $customPayload = array_merge($janeRetrieved->broadcastWith(), ['socket' => null]);
+                ->with(m::on(function ($arg) use ($event) {
+                    $customPayload = array_merge($event->broadcastWith(), ['socket' => null]);
 
                     return $arg['Entries'][0]['Detail'] === json_encode($customPayload)
                         && $arg['Entries'][0]['DetailType'] === UserRetrievedWithCustomPayload::class
@@ -92,40 +85,90 @@ public function it_broadcasts_basic_event_with_action_and_custom_payload()
                 }));
         });
 
-        event($janeRetrieved);
+        event($event);
     }
 
     /** @test */
     public function it_broadcasts_basic_event_to_multiple_channels_as_buses()
     {
-        $jane = $this->createJane();
-
-        $janeRetrieved = new UserRetrievedWithMultipleChannels($jane);
+        $event = new UserRetrievedWithMultipleChannels($this->createJane());
 
-        $this->mockEventBridge(function (MockInterface $eventBridge) use ($janeRetrieved) {
+        $this->mockEventBridge(function (MockInterface $eventBridge) use ($event) {
             $eventBridge
                 ->shouldReceive('putEvents')
                 ->once()
-                ->with(Mockery::on(function ($arg) use ($janeRetrieved) {
-                    return collect($janeRetrieved->broadcastOn())
-                        ->map(function ($channel, $key) use ($arg, $janeRetrieved) {
-                            return $arg['Entries'][$key]['Detail'] === json_encode($janeRetrieved)
+                ->with(m::on(function ($arg) use ($event) {
+                    return collect($event->broadcastOn())
+                        ->map(function ($channel, $key) use ($arg, $event) {
+                            return $arg['Entries'][$key]['Detail'] === json_encode($event)
                                 && $arg['Entries'][$key]['DetailType'] === UserRetrievedWithMultipleChannels::class
                                 && $arg['Entries'][$key]['EventBusName'] === $channel
                                 && $arg['Entries'][$key]['Source'] === 'my-app';
                         })
                         ->filter()
-                        ->count() > 0;
+                        ->count() === 2;
                 }));
         });
 
-        event($janeRetrieved);
+        event($event);
     }
 
-    /**
-     * @return User
-     */
-    protected function createJane()
+    /** @test */
+    public function it_can_use_a_source()
+    {
+        config(['broadcasting.connections.eventbridge.source' => 'some-other-source']);
+
+        $event = new UserRetrievedWithMultipleChannels($this->createJane());
+
+        $this->mockEventBridge(function (MockInterface $eventBridge) use ($event) {
+            $eventBridge
+                ->shouldReceive('putEvents')
+                ->once()
+                ->with(m::on(function ($arg) use ($event) {
+                    return collect($event->broadcastOn())
+                            ->map(function ($channel, $key) use ($arg) {
+                                return $arg['Entries'][$key]['Source'] === 'some-other-source';
+                            })
+                            ->filter()
+                            ->count() > 0;
+                }));
+        });
+
+        event($event);
+    }
+
+    /** @test */
+    public function it_logs_errors_when_events_fail_to_send()
+    {
+        $event = new UserRetrieved($this->createJane());
+
+        $failedEntry = [
+            'ErrorCode' => 'InternalFailure',
+            'ErrorMessage' => 'Something went wrong',
+            'EventId' => $this->faker->uuid,
+        ];
+
+        $this->mockEventBridge(function (MockInterface $eventBridge) use ($failedEntry) {
+            $eventBridge
+                ->shouldReceive('putEvents')
+                ->once()
+                ->andReturn(new Result([
+                    'FailedEntryCount' => 1,
+                    'Entries' => [
+                        $failedEntry,
+                        ['EventId' => $this->faker->uuid],
+                    ],
+                ]));
+        });
+
+        Log::shouldReceive('error')->once()->with('Failed to send events to EventBridge', [
+            'errors' => [$failedEntry],
+        ]);
+
+        event($event);
+    }
+
+    protected function createJane(): User
     {
         return User::create([
             'name' => 'Jane',

diff --git a/tests/Sub/Queue/Jobs/SnsEventDispatcherJobTest.php b/tests/Sub/Queue/Jobs/SnsEventDispatcherJobTest.php
@@ -113,7 +113,7 @@ public function it_will_handle_empty_messages_with_a_subject()
     }
 
     /** @test */
-    public function it_will_not_handle_raw_notification_messages()
+    public function it_will_not_handle_raw_notification_messages_and_release_the_message_onto_the_queue()
     {
         Log::shouldReceive('error')->once()->with(
             m::pattern('/^SqsSnsQueue: Invalid SNS payload/'),
@@ -122,32 +122,60 @@ public function it_will_not_handle_raw_notification_messages()
 
         $this->mockedJobData = $this->mockedRawNotificationMessage()['Messages'][0];
 
-        $this->getJob()->fire();
+        $job = $this->getJob();
+        $job->shouldNotReceive('delete');
+        $job->shouldReceive('release')->once();
+
+        $job->fire();
 
         Event::assertNotDispatched('foo');
     }
 
     /** @test */
-    public function it_will_not_handle_messages_where_the_event_name_to_trigger_cannot_be_resolved()
+    public function it_will_not_handle_messages_where_the_event_name_to_trigger_cannot_be_resolved_and_delete_the_message_from_the_queue()
     {
         $this->mockedJobData = $this->mockedRichNotificationMessage([
             'TopicArn' => '',
             'Subject' => '',
         ])['Messages'][0];
 
-        $this->getJob()->fire();
+        $job = $this->getJob();
+        $job->shouldReceive('delete')->once();
+        $job->shouldNotReceive('release');
+
+        $job->fire();
 
         Event::assertNotDispatched('');
     }
 
+    /** @test */
+    public function it_will_delete_the_message_from_the_queue_when_it_managed_to_dispatch_an_event()
+    {
+        $this->mockedJobData = $this->mockedRichNotificationMessage([
+            'TopicArn' => 'TopicArn:123456',
+        ])['Messages'][0];
+
+        $job = $this->getJob();
+        $job->shouldReceive('delete')->once();
+
+        $job->fire();
+
+        Event::assertDispatched('TopicArn:123456');
+    }
+
+    /** @return SnsEventDispatcherJob|\Mockery\LegacyMockInterface */
     protected function getJob()
     {
-        return new SnsEventDispatcherJob(
+        $mock = m::mock(SnsEventDispatcherJob::class, [
             $this->app,
             m::mock(SqsClient::class),
             $this->mockedJobData,
             'connection-name',
-            'https://sqs.someregion.amazonaws.com/1234567891011/pubsub-events'
-        );
+            'https://sqs.someregion.amazonaws.com/1234567891011/pubsub-events',
+        ])->makePartial();
+
+        $mock->shouldReceive('delete', 'release')->byDefault();
+
+        return $mock;
     }
 }
-Original file line number
+Diff line change
@@ Expand Up / @@ -21,6 +21,8 @@ public function fire() @@
                         'message delivery is disabled for your SQS subscription.', $this->job);
                 }
+                $this->release();
                 return;
             }
@@ Expand All / @@ -30,6 +32,8 @@ public function fire() @@
                     'subject' => $this->snsSubject(),
                 ]);
             }
+            $this->delete();
         }
         /**
@@ Expand Down @@