diff --git a/.env.example b/.env.example index 213a146..94ddbae 100644 --- a/.env.example +++ b/.env.example @@ -34,6 +34,7 @@ VITE_SHOW_LOGOUT="${USE_ZANICHELLI_IDP}" IDP_BASE_URL= IDP_COOKIE_NAME= +SUBSCRIBER_PAUSED_TTL= TELESCOPE_ENABLED=true TELESCOPE_CACHE_WATCHER=true diff --git a/ansible/roles/deploy-buzzer/templates/.env.j2 b/ansible/roles/deploy-buzzer/templates/.env.j2 index c4ee8b9..c4abcd9 100644 --- a/ansible/roles/deploy-buzzer/templates/.env.j2 +++ b/ansible/roles/deploy-buzzer/templates/.env.j2 @@ -36,6 +36,7 @@ VITE_SHOW_LOGOUT="${USE_ZANICHELLI_IDP}" IDP_BASE_URL=https://{{ idp_url }} IDP_COOKIE_NAME={{ idp_cookie_name }} +SUBSCRIBER_PAUSED_TTL=3600 TELESCOPE_ENABLED=false TELESCOPE_CACHE_WATCHER=true diff --git a/app/Events/SendMessageEvent.php b/app/Events/SendMessageEvent.php index 4ec5289..ca79d5e 100644 --- a/app/Events/SendMessageEvent.php +++ b/app/Events/SendMessageEvent.php @@ -15,17 +15,21 @@ class SendMessageEvent public $host; public $channelSubscribe; public $channelPriority; + public $channelName; + public $subscriberName; /** * Create a new event instance. * * @return void */ - public function __construct(Message $message, $host, $channelSubscribe, $channelPriority) + public function __construct(Message $message, $host, $channelSubscribe, $channelPriority, $channelName, $subscriberName) { $this->message = $message; $this->host = $host; $this->channelSubscribe = $channelSubscribe; $this->channelPriority = $channelPriority; + $this->channelName = $channelName; + $this->subscriberName = $subscriberName; } } diff --git a/app/Http/Controllers/ChannelController.php b/app/Http/Controllers/ChannelController.php index 9fd1cf5..9f7a3df 100644 --- a/app/Http/Controllers/ChannelController.php +++ b/app/Http/Controllers/ChannelController.php @@ -472,7 +472,7 @@ private function sendMessageTo($request, $channel) foreach ($channel->subscribers as $subscriber) { $relations = $this->channelSubscribeRepository->where($channel->id, $subscriber->id); foreach ($relations as $relation) { - event(new SendMessageEvent($message, $subscriber->host, $relation, $channel->priority)); + event(new SendMessageEvent($message, $subscriber->host, $relation, $channel->priority, $channel?->name ?? "", $subscriber->name)); } } return response()->json([ @@ -498,6 +498,7 @@ private function getCachedChannelByName($channelName) $cacheData = (object)[ "id" => $channel->id, + "name" => $channel->name, "subscribers" => [], "priority" => $channel->priority ]; diff --git a/app/Http/Controllers/PublisherController.php b/app/Http/Controllers/PublisherController.php index 36fea0c..877e4e9 100644 --- a/app/Http/Controllers/PublisherController.php +++ b/app/Http/Controllers/PublisherController.php @@ -231,7 +231,7 @@ public function destroy(int $id) */ public function getPublisher($id) { - $publisher = $this->publisherRepository->find($id);; + $publisher = $this->publisherRepository->find($id); if (!$publisher) { return response()->error404(__('messages.Publisher') . $id); } diff --git a/app/Http/Controllers/SubscriberController.php b/app/Http/Controllers/SubscriberController.php index 9f13691..f64bbdf 100644 --- a/app/Http/Controllers/SubscriberController.php +++ b/app/Http/Controllers/SubscriberController.php @@ -6,6 +6,8 @@ use App\Http\Requests\SubscriberRequest; use App\Http\Resources\SubscriberResource; use App\Http\Repositories\RepositoryInterface; +use Illuminate\Support\Facades\Cache; +use Illuminate\Support\Facades\Config; class SubscriberController extends Controller { @@ -226,10 +228,96 @@ public function destroy(int $id) */ public function getSubscriber($id) { - $subscriber = $this->subscriberRepository->find($id);; + $subscriber = $this->subscriberRepository->find($id); if (!$subscriber) { return response()->error404(__('messages.Subscriber') . $id); } return new SubscriberResource($subscriber); } + + /** + * @OA\Post( + * path="/api/subscribers/{id}/pause", + * summary="Pauses a subscriber from receiving messages", + * tags={"subscribers"}, + * security={{"passport":{}}}, + * description="Use to pauses a subscriber from receiving messages for 1 hour", + * operationId="SubscriberController.pauseSubscriber", + * @OA\Parameter( + * in="path", + * required=true, + * description="Subscriber id", + * name="id", + * @OA\Schema( + * type="integer", + * minimum=1 + * ) + * ), + * @OA\Response( + * response=200, + * ref="#/components/responses/Success200" + * ), + * @OA\Response( + * response=404, + * ref="#/components/responses/Error404" + * ) + * ) + * + * Response to route /api/subscribers/{id}/pause + * + * @param int $id + * + */ + public function pauseSubscriber($id) + { + $subscriber = $this->subscriberRepository->find($id); + if (!$subscriber) { + return response()->error404(__('messages.Subscriber') . $id); + } + Cache::put(config('cache.subscriber_paused_key_prefix') . $id, true, config('cache.subscriber_paused_ttl')); + return response()->success204(); + } + + /** + * @OA\Post( + * path="/api/subscribers/{id}/restore", + * summary="Restore a subscriber from receiving messages", + * tags={"subscribers"}, + * security={{"passport":{}}}, + * description="Use to restore a subscriber from receiving messages", + * operationId="SubscriberController.restoreSubscriber", + * @OA\Parameter( + * in="path", + * required=true, + * description="Subscriber id", + * name="id", + * @OA\Schema( + * type="integer", + * minimum=1 + * ) + * ), + * @OA\Response( + * response=200, + * ref="#/components/responses/Success200" + * ), + * @OA\Response( + * response=404, + * ref="#/components/responses/Error404" + * ) + * ) + * + * Response to route /api/subscribers/{id}/restore + * + * @param int $id + * + */ + public function restoreSubscriber($id) + { + $subscriber = $this->subscriberRepository->find($id); + if (!$subscriber) { + return response()->error404(__('messages.Subscriber') . $id); + } + Cache::forget(Config::get('cache.subscriber_paused_key_prefix') . $id); + return response()->success204(); + } } diff --git a/app/Jobs/SendMessageJob.php b/app/Jobs/SendMessageJob.php index fe52fd5..179d3c9 100644 --- a/app/Jobs/SendMessageJob.php +++ b/app/Jobs/SendMessageJob.php @@ -10,7 +10,8 @@ use Illuminate\Queue\InteractsWithQueue; use Illuminate\Contracts\Queue\ShouldQueue; use Illuminate\Foundation\Bus\Dispatchable; - +use Illuminate\Support\Facades\Log; +use Illuminate\Support\Facades\Cache; class SendMessageJob implements ShouldQueue { @@ -18,7 +19,7 @@ class SendMessageJob implements ShouldQueue private $event; private $body; - + private $startTime; /** * The number of times the job may be attempted. @@ -67,16 +68,43 @@ public function __construct(SendMessageEvent $event) */ public function handle(GuzzleService $guzzleService) { - switch ($this->event->channelSubscribe->authentication) { - case Authentication::BASIC: - $guzzleService->sendWithBasicAuth($this->event); - break; - case Authentication::NONE: - $guzzleService->sendWithoutAuth($this->event); - break; - case Authentication::OAUTH2: - $guzzleService->sendWithOAuth2($this->event); - break; + try { + Log::withContext([ + "subscriber" => $this->event->subscriberName, + "channel" => $this->event->channelName, + "priority" => $this->event->channelSubscribe->channelPriority, + "authentication" => $this->event->channelSubscribe->authentication, + ]); + + $pausedKey = config('cache.subscriber_paused_key_prefix') . $this->event->channelSubscribe->subscriber_id; + if (Cache::has($pausedKey)) { + Log::warning("Message paused"); + $this->fail("Subscriber Paused"); + return; + } + + $this->startTime = microtime(true); + switch ($this->event->channelSubscribe->authentication) { + case Authentication::BASIC: + $guzzleService->sendWithBasicAuth($this->event); + break; + case Authentication::NONE: + $guzzleService->sendWithoutAuth($this->event); + break; + case Authentication::OAUTH2: + $guzzleService->sendWithOAuth2($this->event); + break; + } + Log::info("Message sent successfully", [ + "time" => microtime(true) - $this->startTime, + ]); + } catch (\Exception $e) { + Log::error('Message sending failed', [ + "time" => microtime(true) - $this->startTime, + "attempts" => $this->attempts(), + "exception" => $e->getMessage(), + ]); + throw $e; } } diff --git a/config/cache.php b/config/cache.php index 910bd25..574d6fb 100644 --- a/config/cache.php +++ b/config/cache.php @@ -104,6 +104,8 @@ */ 'prefix' => env('CACHE_PREFIX', Str::slug(env('APP_NAME', 'laravel'), '_') . '_cache:'), - 'channel_key_prefix' => "channel." + 'channel_key_prefix' => "channel.", + 'subscriber_paused_key_prefix' => "subscriber_paused.", + 'subscriber_paused_ttl' => env('SUBSCRIBER_PAUSED_TTL', 3600) ]; diff --git a/resources/js/store/channels.js b/resources/js/store/channels.js index ec4effc..04ed6c8 100644 --- a/resources/js/store/channels.js +++ b/resources/js/store/channels.js @@ -28,7 +28,7 @@ export const useChannelsStore = defineStore("channels", { return { meta, links }; } catch (error) { this.loadingObject = false; - throw error.response.data;; + throw error.response.data; } }, diff --git a/routes/api.php b/routes/api.php index 8787ab0..42c8a78 100644 --- a/routes/api.php +++ b/routes/api.php @@ -26,6 +26,8 @@ Route::get('/', 'SubscriberController@getList'); Route::post('/', 'SubscriberController@store'); Route::get('{id}', 'SubscriberController@getSubscriber')->where('id', '[0-9]+'); + Route::post('{id}/pause', 'SubscriberController@pauseSubscriber')->where('id', '[0-9]+'); + Route::post('{id}/restore', 'SubscriberController@restoreSubscriber')->where('id', '[0-9]+'); Route::delete('{id}', 'SubscriberController@destroy')->where('id', '[0-9]+'); Route::get('{id}/channels', 'ChannelSubscribeController@getChannelSubscribe')->where('id', '[0-9]+'); diff --git a/tests/Feature/FailedJobTest.php b/tests/Feature/FailedJobTest.php index c156e12..ad91b2f 100644 --- a/tests/Feature/FailedJobTest.php +++ b/tests/Feature/FailedJobTest.php @@ -39,7 +39,7 @@ private function getJsonFragment(?FailedJob $failedJob = null, ?ChannelSubscribe public function testSuccesfullyListFailedJob() { $channelSubscribe = factory(ChannelSubscribe::class)->create(); - $sendMessageJob = new SendMessageJob(new SendMessageEvent(new Message('messaggio'), 'host', $channelSubscribe, 'highpriority')); + $sendMessageJob = new SendMessageJob(new SendMessageEvent(new Message('messaggio'), 'host', $channelSubscribe, 'highpriority', 'channelName', 'publisherName')); $failedJob = factory(FailedJob::class)->make([ 'payload' => json_encode(["data" => ["command" => serialize($sendMessageJob)]]), diff --git a/tests/Feature/SubscriberTest.php b/tests/Feature/SubscriberTest.php index f01ca5c..fab3ce6 100644 --- a/tests/Feature/SubscriberTest.php +++ b/tests/Feature/SubscriberTest.php @@ -6,6 +6,9 @@ use App\Models\Subscriber; use App\Http\Repositories\SubscriberRepository; use Tests\TestCaseWithoutMiddleware; +use Illuminate\Http\Response; +use Illuminate\Support\Facades\Cache; +use Illuminate\Support\Facades\Config; class SubscriberTest extends TestCaseWithoutMiddleware { @@ -169,4 +172,68 @@ public function testDestroy() $response = $this->json('DELETE', '/api/subscribers/1'); $response->assertStatus(200); } + + public function testPause() + { + Cache::spy(); + $subscriber = factory(Subscriber::class)->make(); + $subscriber->id = 1; + $mock = Mockery::mock(SubscriberRepository::class)->makePartial() + ->shouldReceive([ + "find" => $subscriber, + ]) + ->withAnyArgs() + ->once() + ->getMock(); + $this->app->instance('App\Http\Repositories\SubscriberRepository', $mock); + $response = $this->json('POST', '/api/subscribers/1/pause'); + $response->assertStatus(Response::HTTP_NO_CONTENT); + Cache::shouldHaveReceived('put')->with(Config::get('cache.subscriber_paused_key_prefix') . 1, true, 3600); + } + + public function testPauseNotFound() + { + $mock = Mockery::mock(SubscriberRepository::class)->makePartial() + ->shouldReceive([ + "find" => null, + ]) + ->withAnyArgs() + ->once() + ->getMock(); + $this->app->instance('App\Http\Repositories\SubscriberRepository', $mock); + $response = $this->json('POST', '/api/subscribers/1/pause'); + $response->assertStatus(Response::HTTP_NOT_FOUND); + } + + public function testRestore() + { + Cache::spy(); + $subscriber = factory(Subscriber::class)->make(); + $subscriber->id = 1; + $mock = Mockery::mock(SubscriberRepository::class)->makePartial() + ->shouldReceive([ + "find" => $subscriber, + ]) + ->withAnyArgs() + ->once() + ->getMock(); + $this->app->instance('App\Http\Repositories\SubscriberRepository', $mock); + $response = $this->json('POST', '/api/subscribers/1/restore'); + $response->assertStatus(Response::HTTP_NO_CONTENT); + Cache::shouldHaveReceived('forget')->with(Config::get('cache.subscriber_paused_key_prefix') . 1); + } + + public function testRestoreNotFound() + { + $mock = Mockery::mock(SubscriberRepository::class)->makePartial() + ->shouldReceive([ + "find" => null, + ]) + ->withAnyArgs() + ->once() + ->getMock(); + $this->app->instance('App\Http\Repositories\SubscriberRepository', $mock); + $response = $this->json('POST', '/api/subscribers/1/restore'); + $response->assertStatus(Response::HTTP_NOT_FOUND); + } } diff --git a/tests/Integration/MessageTest.php b/tests/Integration/MessageTest.php index b96d7ff..a3b0f38 100644 --- a/tests/Integration/MessageTest.php +++ b/tests/Integration/MessageTest.php @@ -2,7 +2,9 @@ namespace Tests\Integration; +use App\Events\SendMessageEvent; use Mockery; +use App\Models\Message; use App\Models\Channel; use App\Models\Publisher; use Illuminate\Support\Str; @@ -13,6 +15,8 @@ use Illuminate\Foundation\Testing\TestCase; use Illuminate\Support\Facades\Queue; use Tests\CreatesApplication; +use Illuminate\Support\Facades\Cache; +use Illuminate\Support\Facades\Config; class MessageTest extends TestCase { @@ -191,4 +195,17 @@ public function testBasicAuthFailedAuth() $response = $this->json('POST', '/api/sendMessage/' . Str::random(10), [], $authorization); $response->assertStatus(401); } + + public function testSendMessagePaused() + { + $channel = factory(Channel::class)->create(); + $publisher = factory(Publisher::class)->create(['password' => bcrypt(self::PUBLISHER_PASSWORD)]); + factory(ChannelPublish::class)->create(['channel_id' => $channel->id, 'publisher_id' => $publisher->id]); + $channelSubscribe = factory(ChannelSubscribe::class)->create(['channel_id' => $channel->id]); + Cache::put(Config::get('cache.subscriber_paused_key_prefix') . $channelSubscribe->subscriber_id, true, 3600); + $job = (new SendMessageJob(new SendMessageEvent(new Message("hello"), "http://prova.com", $channelSubscribe, "low", "test", "test")))->withFakeQueueInteractions(); + + $job->handle(new GuzzleService()); + $job->assertFailed(); + } }