From b8aa2faa7b6d6da307e2d7d143b86e57568bfbb4 Mon Sep 17 00:00:00 2001 From: Prateek Banga Date: Thu, 7 Dec 2023 11:25:19 +0100 Subject: [PATCH 01/30] adds scheduling commit, still need to resolve pools error --- Dockerfile | 1 + app/config/collections.php | 29 ++++ app/controllers/api/account.php | 2 - app/controllers/api/messaging.php | 119 ++++++++++--- app/controllers/api/teams.php | 1 - app/controllers/shared/api.php | 7 +- bin/schedule-message | 3 + composer.lock | 50 +++--- docker-compose.yml | 27 +++ src/Appwrite/Platform/Services/Tasks.php | 2 + src/Appwrite/Platform/Tasks/Maintenance.php | 2 +- .../Platform/Tasks/ScheduleMessage.php | 164 ++++++++++++++++++ src/Appwrite/Platform/Workers/Deletes.php | 6 +- src/Appwrite/Platform/Workers/Messaging.php | 2 - .../e2e/Services/Messaging/MessagingBase.php | 17 +- 15 files changed, 364 insertions(+), 68 deletions(-) create mode 100644 bin/schedule-message create mode 100644 src/Appwrite/Platform/Tasks/ScheduleMessage.php diff --git a/Dockerfile b/Dockerfile index 059c499bd9..ee9818f390 100755 --- a/Dockerfile +++ b/Dockerfile @@ -80,6 +80,7 @@ RUN chmod +x /usr/local/bin/doctor && \ chmod +x /usr/local/bin/migrate && \ chmod +x /usr/local/bin/realtime && \ chmod +x /usr/local/bin/schedule && \ + chmod +x /usr/local/bin/schedule-message && \ chmod +x /usr/local/bin/sdks && \ chmod +x /usr/local/bin/specs && \ chmod +x /usr/local/bin/ssl && \ diff --git a/app/config/collections.php b/app/config/collections.php index cbaed36f71..913de3e066 100644 --- a/app/config/collections.php +++ b/app/config/collections.php @@ -1593,6 +1593,28 @@ $commonCollections = [ 'array' => false, 'filters' => ['datetime'], ], + [ + '$id' => ID::custom('scheduleInternalId'), + 'type' => Database::VAR_STRING, + 'format' => '', + 'size' => Database::LENGTH_KEY, + 'signed' => true, + 'required' => false, + 'default' => null, + 'array' => false, + 'filters' => [], + ], + [ + '$id' => ID::custom('scheduleId'), + 'type' => Database::VAR_STRING, + 'format' => '', + 'size' => Database::LENGTH_KEY, + 'signed' => true, + 'required' => false, + 'default' => null, + 'array' => false, + 'filters' => [], + ], [ '$id' => ID::custom('deliveredAt'), 'type' => Database::VAR_DATETIME, @@ -4166,6 +4188,13 @@ $consoleCollections = array_merge([ 'lengths' => [], 'orders' => [], ], + [ + '$id' => ID::custom('_key_schedule_resourceType_active_resourceUpdatedAt'), + 'type' => Database::INDEX_KEY, + 'attributes' => ['schedule', 'resourceType', 'active', 'resourceUpdatedAt'], + 'lengths' => [], + 'orders' => [], + ] ], ], diff --git a/app/controllers/api/account.php b/app/controllers/api/account.php index 68e3261a8d..1c32dad629 100644 --- a/app/controllers/api/account.php +++ b/app/controllers/api/account.php @@ -1375,7 +1375,6 @@ App::post('/v1/account/sessions/phone') ->setMessage($messageDoc) ->setRecipients([$phone]) ->setProviderType(MESSAGE_TYPE_SMS) - ->setProject($project) ->trigger(); $queueForEvents->setPayload( @@ -3101,7 +3100,6 @@ App::post('/v1/account/verification/phone') ->setMessage($messageDoc) ->setRecipients([$user->getAttribute('phone')]) ->setProviderType(MESSAGE_TYPE_SMS) - ->setProject($project) ->trigger(); $queueForEvents diff --git a/app/controllers/api/messaging.php b/app/controllers/api/messaging.php index 411243d7b3..88734b7c7c 100644 --- a/app/controllers/api/messaging.php +++ b/app/controllers/api/messaging.php @@ -34,6 +34,7 @@ use Utopia\Validator\Boolean; use Utopia\Validator\JSON; use Utopia\Validator\Text; use MaxMind\Db\Reader; +use Utopia\Database\DateTime; use Utopia\Validator\WhiteList; use function Swoole\Coroutine\batch; @@ -1441,7 +1442,6 @@ App::patch('/v1/messaging/providers/fcm/:providerId') ->label('audits.resource', 'provider/{response.$id}') ->label('event', 'providers.[providerId].update') ->label('scope', 'providers.write') - ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY]) ->label('sdk.namespace', 'messaging') ->label('sdk.method', 'updateFcmProvider') ->label('sdk.description', '/docs/references/messaging/update-fcm-provider.md') @@ -2228,10 +2228,11 @@ App::post('/v1/messaging/messages/email') ->param('scheduledAt', null, new DatetimeValidator(requireDateInFuture: true), 'Scheduled delivery time for message in [ISO 8601](https://www.iso.org/iso-8601-date-and-time-format.html) format. DateTime value must be in future.', true) ->inject('queueForEvents') ->inject('dbForProject') + ->inject('dbForConsole') ->inject('project') ->inject('queueForMessaging') ->inject('response') - ->action(function (string $messageId, string $subject, string $content, array $topics, array $users, array $targets, string $description, string $status, bool $html, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Document $project, Messaging $queueForMessaging, Response $response) { + ->action(function (string $messageId, string $subject, string $content, array $topics, array $users, array $targets, string $description, string $status, bool $html, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Database $dbForConsole, Document $project, Messaging $queueForMessaging, Response $response) { $messageId = $messageId == 'unique()' ? ID::unique() : $messageId; if (\count($topics) === 0 && \count($users) === 0 && \count($targets) === 0) { @@ -2245,6 +2246,7 @@ App::post('/v1/messaging/messages/email') 'users' => $users, 'targets' => $targets, 'description' => $description, + 'scheduledAt' => $scheduledAt, 'data' => [ 'subject' => $subject, 'content' => $content, @@ -2253,11 +2255,24 @@ App::post('/v1/messaging/messages/email') 'status' => $status, ])); - if ($status === 'processing') { + if ($status === 'processing' && $scheduledAt === null) { $queueForMessaging ->setMessageId($message->getId()) - ->setProject($project) ->trigger(); + } else if ($scheduledAt !== null) { + $schedule = $dbForConsole->createDocument('schedules', new Document([ + 'region' => App::getEnv('_APP_REGION', 'default'), + 'resourceType' => 'message', + 'resourceId' => $message->getId(), + 'resourceInternalId' => $message->getInternalId(), + 'resourceUpdatedAt' => DateTime::now(), + 'projectId' => $project->getId(), + 'schedule' => $message->getAttribute('scheduledAt'), + 'active' => $status === 'processing' ? true : false, + ])); + + $message->setAttribute('scheduleId', $schedule->getId()); + $dbForProject->updateDocument('messages', $message->getId(), $message); } $queueForEvents @@ -2292,10 +2307,11 @@ App::post('/v1/messaging/messages/sms') ->param('scheduledAt', null, new DatetimeValidator(requireDateInFuture: true), 'Scheduled delivery time for message in [ISO 8601](https://www.iso.org/iso-8601-date-and-time-format.html) format. DateTime value must be in future.', true) ->inject('queueForEvents') ->inject('dbForProject') + ->inject('dbForConsole') ->inject('project') ->inject('queueForMessaging') ->inject('response') - ->action(function (string $messageId, string $content, array $topics, array $users, array $targets, string $description, string $status, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Document $project, Messaging $queueForMessaging, Response $response) { + ->action(function (string $messageId, string $content, array $topics, array $users, array $targets, string $description, string $status, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Database $dbForConsole, Document $project, Messaging $queueForMessaging, Response $response) { $messageId = $messageId == 'unique()' ? ID::unique() : $messageId; if (\count($topics) === 0 && \count($users) === 0 && \count($targets) === 0) { @@ -2315,11 +2331,24 @@ App::post('/v1/messaging/messages/sms') 'status' => $status, ])); - if ($status === 'processing') { + if ($status === 'processing' && $scheduledAt === null) { $queueForMessaging ->setMessageId($message->getId()) - ->setProject($project) ->trigger(); + } else if ($status === 'processing' && $scheduledAt !== null) { + $schedule = $dbForConsole->createDocument('schedules', new Document([ + 'region' => App::getEnv('_APP_REGION', 'default'), + 'resourceType' => 'message', + 'resourceId' => $message->getId(), + 'resourceInternalId' => $message->getInternalId(), + 'resourceUpdatedAt' => DateTime::now(), + 'projectId' => $project->getId(), + 'schedule' => $message->getAttribute('scheduledAt'), + 'active' => $status === 'processing' ? true : false, + ])); + + $message->setAttribute('scheduleId', $schedule->getId()); + $dbForProject->updateDocument('messages', $message->getId(), $message); } $queueForEvents @@ -2362,10 +2391,11 @@ App::post('/v1/messaging/messages/push') ->param('scheduledAt', null, new DatetimeValidator(requireDateInFuture: true), 'Scheduled delivery time for message in [ISO 8601](https://www.iso.org/iso-8601-date-and-time-format.html) format. DateTime value must be in future.', true) ->inject('queueForEvents') ->inject('dbForProject') + ->inject('dbForConsole') ->inject('project') ->inject('queueForMessaging') ->inject('response') - ->action(function (string $messageId, string $title, string $body, array $topics, array $users, array $targets, string $description, ?array $data, string $action, string $icon, string $sound, string $color, string $tag, string $badge, string $status, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Document $project, Messaging $queueForMessaging, Response $response) { + ->action(function (string $messageId, string $title, string $body, array $topics, array $users, array $targets, string $description, ?array $data, string $action, string $icon, string $sound, string $color, string $tag, string $badge, string $status, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Database $dbForConsole, Document $project, Messaging $queueForMessaging, Response $response) { $messageId = $messageId == 'unique()' ? ID::unique() : $messageId; if (\count($topics) === 0 && \count($users) === 0 && \count($targets) === 0) { @@ -2394,11 +2424,24 @@ App::post('/v1/messaging/messages/push') 'status' => $status, ])); - if ($status === 'processing') { + if ($status === 'processing' && $scheduledAt === null) { $queueForMessaging ->setMessageId($message->getId()) - ->setProject($project) ->trigger(); + } else if ($status === 'processing' && $scheduledAt !== null) { + $schedule = $dbForConsole->createDocument('schedules', new Document([ + 'region' => App::getEnv('_APP_REGION', 'default'), + 'resourceType' => 'message', + 'resourceId' => $message->getId(), + 'resourceInternalId' => $message->getInternalId(), + 'resourceUpdatedAt' => DateTime::now(), + 'projectId' => $project->getId(), + 'schedule' => $message->getAttribute('scheduledAt'), + 'active' => $status === 'processing' ? true : false, + ])); + + $message->setAttribute('scheduleId', $schedule->getId()); + $dbForProject->updateDocument('messages', $message->getId(), $message); } $queueForEvents @@ -2586,10 +2629,11 @@ App::patch('/v1/messaging/messages/email/:messageId') ->param('scheduledAt', null, new DatetimeValidator(requireDateInFuture: true), 'Scheduled delivery time for message in [ISO 8601](https://www.iso.org/iso-8601-date-and-time-format.html) format. DateTime value must be in future.', true) ->inject('queueForEvents') ->inject('dbForProject') + ->inject('dbForConsole') ->inject('project') ->inject('queueForMessaging') ->inject('response') - ->action(function (string $messageId, ?array $topics, ?array $users, ?array $targets, string $subject, string $description, string $content, string $status, bool $html, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Document $project, Messaging $queueForMessaging, Response $response) { + ->action(function (string $messageId, ?array $topics, ?array $users, ?array $targets, string $subject, string $description, string $content, string $status, bool $html, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Database $dbForConsole, Document $project, Messaging $queueForMessaging, Response $response) { $message = $dbForProject->getDocument('messages', $messageId); if ($message->isEmpty()) { @@ -2642,14 +2686,25 @@ App::patch('/v1/messaging/messages/email/:messageId') if (!is_null($scheduledAt)) { $message->setAttribute('scheduledAt', $scheduledAt); + + $schedule = $dbForConsole->getDocument('schedules', $message->getAttribute('scheduleId')); + + $schedule + ->setAttribute('resourceUpdatedAt', DateTime::now()) + ->setAttribute('schedule', $message->getAttribute('schedule')); + + if ($message->getAttribute('status') === 'processing') { + $schedule->setAttribute('active', true); + } + + $dbForConsole->updateDocument('schedules', $schedule->getId(), $schedule); } $message = $dbForProject->updateDocument('messages', $message->getId(), $message); - if ($status === 'processing') { + if ($status === 'processing' && \is_null($message->getAttribute('scheduledAt'))) { $queueForMessaging ->setMessageId($message->getId()) - ->setProject($project) ->trigger(); } @@ -2684,10 +2739,11 @@ App::patch('/v1/messaging/messages/sms/:messageId') ->param('scheduledAt', null, new DatetimeValidator(requireDateInFuture: true), 'Scheduled delivery time for message in [ISO 8601](https://www.iso.org/iso-8601-date-and-time-format.html) format. DateTime value must be in future.', true) ->inject('queueForEvents') ->inject('dbForProject') + ->inject('dbForConsole') ->inject('project') ->inject('queueForMessaging') ->inject('response') - ->action(function (string $messageId, ?array $topics, ?array $users, ?array $targets, string $description, string $content, string $status, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Document $project, Messaging $queueForMessaging, Response $response) { + ->action(function (string $messageId, ?array $topics, ?array $users, ?array $targets, string $description, string $content, string $status, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Database $dbForConsole, Document $project, Messaging $queueForMessaging, Response $response) { $message = $dbForProject->getDocument('messages', $messageId); if ($message->isEmpty()) { @@ -2732,14 +2788,25 @@ App::patch('/v1/messaging/messages/sms/:messageId') if (!is_null($scheduledAt)) { $message->setAttribute('scheduledAt', $scheduledAt); + + $schedule = $dbForConsole->getDocument('schedules', $message->getAttribute('scheduleId')); + + $schedule + ->setAttribute('resourceUpdatedAt', DateTime::now()) + ->setAttribute('schedule', $message->getAttribute('schedule')); + + if ($message->getAttribute('status') === 'processing') { + $schedule->setAttribute('active', true); + } + + $dbForConsole->updateDocument('schedules', $schedule->getId(), $schedule); } $message = $dbForProject->updateDocument('messages', $message->getId(), $message); - if ($status === 'processing') { + if ($status === 'processing' && \is_null($message->getAttribute('scheduledAt'))) { $queueForMessaging ->setMessageId($message->getId()) - ->setProject($project) ->trigger(); } @@ -2781,10 +2848,11 @@ App::patch('/v1/messaging/messages/push/:messageId') ->param('scheduledAt', null, new DatetimeValidator(requireDateInFuture: true), 'Scheduled delivery time for message in [ISO 8601](https://www.iso.org/iso-8601-date-and-time-format.html) format. DateTime value must be in future.', true) ->inject('queueForEvents') ->inject('dbForProject') + ->inject('dbForConsole') ->inject('project') ->inject('queueForMessaging') ->inject('response') - ->action(function (string $messageId, ?array $topics, ?array $users, ?array $targets, string $description, string $title, string $body, ?array $data, string $action, string $icon, string $sound, string $color, string $tag, string $badge, string $status, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Document $project, Messaging $queueForMessaging, Response $response) { + ->action(function (string $messageId, ?array $topics, ?array $users, ?array $targets, string $description, string $title, string $body, ?array $data, string $action, string $icon, string $sound, string $color, string $tag, string $badge, string $status, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Database $dbForConsole, Document $project, Messaging $queueForMessaging, Response $response) { $message = $dbForProject->getDocument('messages', $messageId); if ($message->isEmpty()) { @@ -2861,16 +2929,27 @@ App::patch('/v1/messaging/messages/push/:messageId') if (!is_null($scheduledAt)) { $message->setAttribute('scheduledAt', $scheduledAt); + + $schedule = $dbForConsole->getDocument('schedules', $message->getAttribute('scheduleId')); + + $schedule + ->setAttribute('resourceUpdatedAt', DateTime::now()) + ->setAttribute('schedule', $message->getAttribute('schedule')); + + if ($message->getAttribute('status') === 'processing') { + $schedule->setAttribute('active', true); + } + + $dbForConsole->updateDocument('schedules', $schedule->getId(), $schedule); } $message = $dbForProject->updateDocument('messages', $message->getId(), $message); - if ($status === 'processing') { + if ($status === 'processing' && \is_null($message->getAttribute('scheduledAt'))) { $queueForMessaging ->setMessageId($message->getId()) - ->setProject($project) ->trigger(); - } + } $queueForEvents ->setParam('messageId', $message->getId()); diff --git a/app/controllers/api/teams.php b/app/controllers/api/teams.php index 2ba27efcb3..14ae4723bc 100644 --- a/app/controllers/api/teams.php +++ b/app/controllers/api/teams.php @@ -653,7 +653,6 @@ App::post('/v1/teams/:teamId/memberships') ->setMessage($messageDoc) ->setRecipients([$phone]) ->setProviderType('SMS') - ->setProject($project) ->trigger(); } } diff --git a/app/controllers/shared/api.php b/app/controllers/shared/api.php index b37d76a816..a988cb7e68 100644 --- a/app/controllers/shared/api.php +++ b/app/controllers/shared/api.php @@ -7,6 +7,7 @@ use Appwrite\Event\Delete; use Appwrite\Event\Event; use Appwrite\Event\Func; use Appwrite\Event\Mail; +use Appwrite\Event\Messaging; use Appwrite\Extend\Exception; use Appwrite\Messaging\Adapter\Realtime; use Appwrite\Usage\Stats; @@ -97,6 +98,7 @@ App::init() ->inject('project') ->inject('user') ->inject('queueForEvents') + ->inject('queueForMessaging') ->inject('queueForAudits') ->inject('queueForDeletes') ->inject('queueForDatabase') @@ -104,7 +106,7 @@ App::init() ->inject('mode') ->inject('queueForMails') ->inject('usage') - ->action(function (App $utopia, Request $request, Response $response, Document $project, Document $user, Event $queueForEvents, Audit $queueForAudits, Delete $queueForDeletes, EventDatabase $queueForDatabase, Database $dbForProject, string $mode, Mail $queueForMails, Stats $usage) use ($databaseListener) { + ->action(function (App $utopia, Request $request, Response $response, Document $project, Document $user, Event $queueForEvents, Messaging $queueForMessaging, Audit $queueForAudits, Delete $queueForDeletes, EventDatabase $queueForDatabase, Database $dbForProject, string $mode, Mail $queueForMails, Stats $usage) use ($databaseListener) { $route = $utopia->getRoute(); @@ -178,6 +180,9 @@ App::init() ->setProject($project) ->setUser($user); + $queueForMessaging + ->setProject($project); + $queueForAudits ->setMode($mode) ->setUserAgent($request->getUserAgent('')) diff --git a/bin/schedule-message b/bin/schedule-message new file mode 100644 index 0000000000..62e0fdbe6e --- /dev/null +++ b/bin/schedule-message @@ -0,0 +1,3 @@ +#!/bin/sh + +php /usr/src/code/app/cli.php schedule-message $@ \ No newline at end of file diff --git a/composer.lock b/composer.lock index 16f44a6357..88a037b239 100644 --- a/composer.lock +++ b/composer.lock @@ -402,16 +402,16 @@ }, { "name": "guzzlehttp/guzzle", - "version": "7.8.0", + "version": "7.8.1", "source": { "type": "git", "url": "https://github.com/guzzle/guzzle.git", - "reference": "1110f66a6530a40fe7aea0378fe608ee2b2248f9" + "reference": "41042bc7ab002487b876a0683fc8dce04ddce104" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/guzzle/guzzle/zipball/1110f66a6530a40fe7aea0378fe608ee2b2248f9", - "reference": "1110f66a6530a40fe7aea0378fe608ee2b2248f9", + "url": "https://api.github.com/repos/guzzle/guzzle/zipball/41042bc7ab002487b876a0683fc8dce04ddce104", + "reference": "41042bc7ab002487b876a0683fc8dce04ddce104", "shasum": "" }, "require": { @@ -426,11 +426,11 @@ "psr/http-client-implementation": "1.0" }, "require-dev": { - "bamarni/composer-bin-plugin": "^1.8.1", + "bamarni/composer-bin-plugin": "^1.8.2", "ext-curl": "*", "php-http/client-integration-tests": "dev-master#2c025848417c1135031fdf9c728ee53d0a7ceaee as 3.0.999", "php-http/message-factory": "^1.1", - "phpunit/phpunit": "^8.5.29 || ^9.5.23", + "phpunit/phpunit": "^8.5.36 || ^9.6.15", "psr/log": "^1.1 || ^2.0 || ^3.0" }, "suggest": { @@ -508,7 +508,7 @@ ], "support": { "issues": "https://github.com/guzzle/guzzle/issues", - "source": "https://github.com/guzzle/guzzle/tree/7.8.0" + "source": "https://github.com/guzzle/guzzle/tree/7.8.1" }, "funding": [ { @@ -524,28 +524,28 @@ "type": "tidelift" } ], - "time": "2023-08-27T10:20:53+00:00" + "time": "2023-12-03T20:35:24+00:00" }, { "name": "guzzlehttp/promises", - "version": "2.0.1", + "version": "2.0.2", "source": { "type": "git", "url": "https://github.com/guzzle/promises.git", - "reference": "111166291a0f8130081195ac4556a5587d7f1b5d" + "reference": "bbff78d96034045e58e13dedd6ad91b5d1253223" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/guzzle/promises/zipball/111166291a0f8130081195ac4556a5587d7f1b5d", - "reference": "111166291a0f8130081195ac4556a5587d7f1b5d", + "url": "https://api.github.com/repos/guzzle/promises/zipball/bbff78d96034045e58e13dedd6ad91b5d1253223", + "reference": "bbff78d96034045e58e13dedd6ad91b5d1253223", "shasum": "" }, "require": { "php": "^7.2.5 || ^8.0" }, "require-dev": { - "bamarni/composer-bin-plugin": "^1.8.1", - "phpunit/phpunit": "^8.5.29 || ^9.5.23" + "bamarni/composer-bin-plugin": "^1.8.2", + "phpunit/phpunit": "^8.5.36 || ^9.6.15" }, "type": "library", "extra": { @@ -591,7 +591,7 @@ ], "support": { "issues": "https://github.com/guzzle/promises/issues", - "source": "https://github.com/guzzle/promises/tree/2.0.1" + "source": "https://github.com/guzzle/promises/tree/2.0.2" }, "funding": [ { @@ -607,20 +607,20 @@ "type": "tidelift" } ], - "time": "2023-08-03T15:11:55+00:00" + "time": "2023-12-03T20:19:20+00:00" }, { "name": "guzzlehttp/psr7", - "version": "2.6.1", + "version": "2.6.2", "source": { "type": "git", "url": "https://github.com/guzzle/psr7.git", - "reference": "be45764272e8873c72dbe3d2edcfdfcc3bc9f727" + "reference": "45b30f99ac27b5ca93cb4831afe16285f57b8221" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/guzzle/psr7/zipball/be45764272e8873c72dbe3d2edcfdfcc3bc9f727", - "reference": "be45764272e8873c72dbe3d2edcfdfcc3bc9f727", + "url": "https://api.github.com/repos/guzzle/psr7/zipball/45b30f99ac27b5ca93cb4831afe16285f57b8221", + "reference": "45b30f99ac27b5ca93cb4831afe16285f57b8221", "shasum": "" }, "require": { @@ -634,9 +634,9 @@ "psr/http-message-implementation": "1.0" }, "require-dev": { - "bamarni/composer-bin-plugin": "^1.8.1", + "bamarni/composer-bin-plugin": "^1.8.2", "http-interop/http-factory-tests": "^0.9", - "phpunit/phpunit": "^8.5.29 || ^9.5.23" + "phpunit/phpunit": "^8.5.36 || ^9.6.15" }, "suggest": { "laminas/laminas-httphandlerrunner": "Emit PSR-7 responses" @@ -707,7 +707,7 @@ ], "support": { "issues": "https://github.com/guzzle/psr7/issues", - "source": "https://github.com/guzzle/psr7/tree/2.6.1" + "source": "https://github.com/guzzle/psr7/tree/2.6.2" }, "funding": [ { @@ -723,7 +723,7 @@ "type": "tidelift" } ], - "time": "2023-08-27T10:13:57+00:00" + "time": "2023-12-03T20:05:35+00:00" }, { "name": "influxdb/influxdb-php", @@ -5823,5 +5823,5 @@ "platform-overrides": { "php": "8.0" }, - "plugin-api-version": "2.3.0" + "plugin-api-version": "2.6.0" } diff --git a/docker-compose.yml b/docker-compose.yml index a570c5b619..d51dffe479 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -717,6 +717,33 @@ services: - _APP_DB_USER - _APP_DB_PASS + appwrite-schedule-message: + entrypoint: schedule-message + <<: *x-logging + container_name: appwrite-schedule-message + image: appwrite-dev + networks: + - appwrite + volumes: + - ./app:/usr/src/code/app + - ./src:/usr/src/code/src + depends_on: + - mariadb + - redis + environment: + - _APP_ENV + - _APP_WORKER_PER_CORE + - _APP_OPENSSL_KEY_V1 + - _APP_REDIS_HOST + - _APP_REDIS_PORT + - _APP_REDIS_USER + - _APP_REDIS_PASS + - _APP_DB_HOST + - _APP_DB_PORT + - _APP_DB_SCHEMA + - _APP_DB_USER + - _APP_DB_PASS + appwrite-assistant: container_name: appwrite-assistant image: appwrite/assistant:0.2.2 diff --git a/src/Appwrite/Platform/Services/Tasks.php b/src/Appwrite/Platform/Services/Tasks.php index 28d7046dd1..29e86b8c78 100644 --- a/src/Appwrite/Platform/Services/Tasks.php +++ b/src/Appwrite/Platform/Services/Tasks.php @@ -20,6 +20,7 @@ use Appwrite\Platform\Tasks\CalcTierStats; use Appwrite\Platform\Tasks\Upgrade; use Appwrite\Platform\Tasks\DeleteOrphanedProjects; use Appwrite\Platform\Tasks\PatchRecreateRepositoriesDocuments; +use Appwrite\Platform\Tasks\ScheduleMessage; class Tasks extends Service { @@ -36,6 +37,7 @@ class Tasks extends Service ->addAction(Install::getName(), new Install()) ->addAction(Upgrade::getName(), new Upgrade()) ->addAction(Maintenance::getName(), new Maintenance()) + ->addAction(ScheduleMessage::getName(), new ScheduleMessage()) ->addAction(Schedule::getName(), new Schedule()) ->addAction(Migrate::getName(), new Migrate()) ->addAction(SDKs::getName(), new SDKs()) diff --git a/src/Appwrite/Platform/Tasks/Maintenance.php b/src/Appwrite/Platform/Tasks/Maintenance.php index 82a62ffed1..c9928dd163 100644 --- a/src/Appwrite/Platform/Tasks/Maintenance.php +++ b/src/Appwrite/Platform/Tasks/Maintenance.php @@ -85,7 +85,7 @@ class Maintenance extends Action private function notifyDeleteUsageStats(int $usageStatsRetentionHourly, Delete $queueForDeletes): void { - ($queueForDeletes) + ($queueFor) ->setType(DELETE_TYPE_USAGE) ->setUsageRetentionHourlyDateTime(DateTime::addSeconds(new \DateTime(), -1 * $usageStatsRetentionHourly)) ->trigger(); diff --git a/src/Appwrite/Platform/Tasks/ScheduleMessage.php b/src/Appwrite/Platform/Tasks/ScheduleMessage.php new file mode 100644 index 0000000000..cd29a1dcf8 --- /dev/null +++ b/src/Appwrite/Platform/Tasks/ScheduleMessage.php @@ -0,0 +1,164 @@ +desc('Execute functions scheduled in Appwrite') + ->inject('pools') + ->inject('dbForConsole') + ->inject('getProjectDB') + ->callback(fn (Group $pools, Database $dbForConsole, callable $getProjectDB) => $this->action($pools, $dbForConsole, $getProjectDB)); + } + + /** + * 1. Load all documents from 'schedules' collection to create local copy + * 2. Create timer that sync all changes from 'schedules' collection to local copy. Only reading changes thanks to 'resourceUpdatedAt' attribute + * 3. Create timer that prepares coroutines for soon-to-execute schedules. When it's ready, coroutime sleeps until exact time before sending request to worker. + */ + public function action(Group $pools, Database $dbForConsole, callable $getProjectDB): void + { + Console::title('Scheduler V1'); + Console::success(APP_NAME . ' Scheduler v1 has started'); + + $schedules = []; // Local copy of 'schedules' collection + $lastSyncUpdate = DateTime::now(); + + $limit = 10000; + $sum = $limit; + $total = 0; + $loadStart = \microtime(true); + $latestDocument = null; + + while ($sum === $limit) { + $paginationQueries = [Query::limit($limit)]; + if ($latestDocument !== null) { + $paginationQueries[] = Query::cursorAfter($latestDocument); + } + try{ + + + $results = $dbForConsole->find('schedules', \array_merge($paginationQueries, [ + Query::lessThanEqual('schedule', DateTime::formatTz(DateTime::now())), + Query::equal('resourceType', ['message']), + Query::equal('active', [true]), + ])); + } catch (\Exception $e) { + var_dump($e->getTraceAsString()); + } + + $sum = count($results); + $total = $total + $sum; + foreach($results as $schedule) { + $schedules[$schedule->getId()] = $schedule; + } + + $latestDocument = !empty(array_key_last($results)) ? $results[array_key_last($results)] : null; + } + + $pools->reclaim(); + + Console::success("{$total} message were loaded in " . (microtime(true) - $loadStart) . " seconds"); + + Console::success("Starting timers at " . DateTime::now()); + + run( + function () use ($dbForConsole, &$schedules, &$lastSyncUpdate, $pools) { + /** + * The timer synchronize $schedules copy with database collection. + */ + Timer::tick(self::MESSAGE_UPDATE_TIMER * 1000, function () use ($dbForConsole, &$schedules, &$lastSyncUpdate, $pools) { + $time = DateTime::now(); + $timerStart = \microtime(true); + + $limit = 1000; + $sum = $limit; + $total = 0; + $latestDocument = null; + + Console::log("Sync tick: Running at $time"); + + while ($sum === $limit) { + $paginationQueries = [Query::limit($limit)]; + if ($latestDocument !== null) { + $paginationQueries[] = Query::cursorAfter($latestDocument); + } + $results = $dbForConsole->find('schedules', \array_merge($paginationQueries, [ + Query::lessThanEqual('schedule', DateTime::formatTz(DateTime::now())), + Query::equal('resourceType', ['message']), + Query::equal('active', [true]), + ])); + $sum = \count($results); + $total = $total + $sum; + foreach ($results as $schedule) { + $schedules[$schedule->getId()] = $schedule; + } + + $latestDocument = !empty(array_key_last($results)) ? $results[array_key_last($results)] : null; + } + + $lastSyncUpdate = $time; + $timerEnd = \microtime(true); + + $pools->reclaim(); + + Console::log("Sync tick: {$total} schedules were updated in " . ($timerEnd - $timerStart) . " seconds"); + }); + + /** + * The timer to prepare soon-to-execute schedules. + */ + $enqueueMessages = function () use (&$schedules, $pools, $dbForConsole) { + foreach ($schedules as $scheduleId => $schedule) { + \go(function () use ($schedules, $schedule, $pools, $dbForConsole) { + $queue = $pools->get('queue')->pop(); + $connection = $queue->getResource(); + $queueForMessaging = new Messaging($connection); + $queueForDeletes = new Delete($connection); + $project = $dbForConsole->getDocument('projects', $schedule->getAttribute('projectId')); + $queueForMessaging + ->setMessageId($schedule->getAttribute('resourceId')) + ->setProject($project) + ->trigger(); + $schedule->setAttribute('active', false); + $dbForConsole->updateDocument('schedules', $schedule->getId(), $schedule); + + $queueForDeletes + ->setType(DELETE_TYPE_SCHEDULES) + ->setDocument($schedule); + + $queue->reclaim(); + unset($schedules[$schedule->getId()]); + }); + } + }; + + Timer::tick(self::MESSAGE_ENQUEUE_TIMER * 1000, fn () => $enqueueMessages()); + $enqueueMessages(); + } + ); + } +} diff --git a/src/Appwrite/Platform/Workers/Deletes.php b/src/Appwrite/Platform/Workers/Deletes.php index 74365bad86..6bc5db42ff 100644 --- a/src/Appwrite/Platform/Workers/Deletes.php +++ b/src/Appwrite/Platform/Workers/Deletes.php @@ -148,7 +148,7 @@ class Deletes extends Action $this->deleteCacheByDate($project, $getProjectDB, $datetime); break; case DELETE_TYPE_SCHEDULES: - $this->deleteSchedules($dbForConsole, $getProjectDB, $datetime); + $this->deleteSchedules($dbForConsole, $getProjectDB, $datetime, $document); break; case DELETE_TYPE_TOPIC: $this->deleteTopic($project, $getProjectDB, $document); @@ -167,13 +167,13 @@ class Deletes extends Action * @throws Authorization * @throws Throwable */ - private function deleteSchedules(Database $dbForConsole, callable $getProjectDB, string $datetime): void + private function deleteSchedules(Database $dbForConsole, callable $getProjectDB, string $datetime, ?Document $document = null): void { $this->listByGroup( 'schedules', [ Query::equal('region', [App::getEnv('_APP_REGION', 'default')]), - Query::equal('resourceType', ['function']), + Query::equal('resourceType', [$document ?? $document->getAttribute('resourceType')]), Query::lessThanEqual('resourceUpdatedAt', $datetime), Query::equal('active', [false]), ], diff --git a/src/Appwrite/Platform/Workers/Messaging.php b/src/Appwrite/Platform/Workers/Messaging.php index 36647e9b7a..c5a6f90192 100644 --- a/src/Appwrite/Platform/Workers/Messaging.php +++ b/src/Appwrite/Platform/Workers/Messaging.php @@ -106,7 +106,6 @@ class Messaging extends Action $targets = $dbForProject->find('targets', [Query::equal('$id', $targetsId)]); $recipients = \array_merge($recipients, $targets); } - $primaryProvider = $dbForProject->findOne('providers', [ Query::equal('enabled', [true]), Query::equal('type', [$recipients[0]->getAttribute('providerType')]), @@ -155,7 +154,6 @@ class Messaging extends Action $providers[] = $provider; $identifiers = $identifiersByProviderId[$providerId]; - $adapter = match ($provider->getAttribute('type')) { MESSAGE_TYPE_SMS => $this->sms($provider), MESSAGE_TYPE_PUSH => $this->push($provider), diff --git a/tests/e2e/Services/Messaging/MessagingBase.php b/tests/e2e/Services/Messaging/MessagingBase.php index 690e503e77..1ae4e4fecb 100644 --- a/tests/e2e/Services/Messaging/MessagingBase.php +++ b/tests/e2e/Services/Messaging/MessagingBase.php @@ -572,7 +572,8 @@ trait MessagingBase 'apiKey' => $apiKey, 'domain' => $domain, 'isEuRegion' => filter_var($isEuRegion, FILTER_VALIDATE_BOOLEAN), - 'from' => $from + 'from' => $from, + 'enabled' => true, ]); $this->assertEquals(201, $provider['headers']['status-code']); @@ -605,18 +606,8 @@ trait MessagingBase $this->assertEquals(201, $user['headers']['status-code']); // Create Target - $target = $this->client->call(Client::METHOD_POST, '/users/' . $user['body']['$id'] . '/targets', [ - 'content-type' => 'application/json', - 'x-appwrite-project' => $this->getProject()['$id'], - 'x-appwrite-key' => $this->getProject()['apiKey'], - ], [ - 'targetId' => ID::unique(), - 'providerType' => 'email', - 'providerId' => $provider['body']['$id'], - 'identifier' => $to, - ]); + $target = $user['body']['targets'][0]; - $this->assertEquals(201, $target['headers']['status-code']); // Create Subscriber $subscriber = $this->client->call(Client::METHOD_POST, '/messaging/topics/' . $topic['body']['$id'] . '/subscribers', \array_merge([ @@ -624,7 +615,7 @@ trait MessagingBase 'x-appwrite-project' => $this->getProject()['$id'], ], $this->getHeaders()), [ 'subscriberId' => ID::unique(), - 'targetId' => $target['body']['$id'], + 'targetId' => $target['$id'], ]); $this->assertEquals(201, $subscriber['headers']['status-code']); From 6de26597934c9497f5f2bb9487206b54b9a5bea1 Mon Sep 17 00:00:00 2001 From: Prateek Banga Date: Mon, 18 Dec 2023 15:27:09 +0530 Subject: [PATCH 02/30] adds mock providers in project for benchmarking, fixes bug in scheduling removes scheduling from array in schedulemessage task --- app/controllers/api/projects.php | 48 +++++++++++++++++++ .../Platform/Tasks/ScheduleMessage.php | 2 +- 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/app/controllers/api/projects.php b/app/controllers/api/projects.php index fe441e0e8c..38d23b2d2d 100644 --- a/app/controllers/api/projects.php +++ b/app/controllers/api/projects.php @@ -216,6 +216,54 @@ App::post('/v1/projects') } $dbForProject->createCollection($key, $attributes, $indexes); } + $emailProvider = new Document([ + '$id' => ID::custom('mock-email-provider'), + 'name' => 'mock', + 'provider' => 'mock', + 'type' => MESSAGE_TYPE_EMAIL, + 'enabled' => true, + 'credentials' => [ + 'username' => 'username', + 'password' => 'password' + ], + 'options' => [ + 'from' => 'sender-email' + ], + ]); + $smsProvider = new Document([ + '$id' => ID::custom('mock-sms-provider'), + 'name' => 'mock', + 'provider' => 'mock', + 'type' => MESSAGE_TYPE_SMS, + 'enabled' => true, + 'credentials' => [ + 'username' => 'username', + 'password' => 'password' + ], + 'options' => [ + 'from' => 'sender-email' + ], + ]); + $pushProvider = new Document([ + '$id' => ID::custom('mock-push-provider'), + 'name' => 'mock', + 'provider' => 'mock', + 'type' => MESSAGE_TYPE_PUSH, + 'enabled' => true, + 'credentials' => [ + 'username' => 'username', + 'password' => 'password' + ], + 'options' => [ + 'from' => 'sender-email' + ], + ]); + $dbForProject->createDocument('providers', $emailProvider); + + + $dbForProject->createDocument('providers', $smsProvider); + + $dbForProject->createDocument('providers', $pushProvider); $response ->setStatusCode(Response::STATUS_CODE_CREATED) diff --git a/src/Appwrite/Platform/Tasks/ScheduleMessage.php b/src/Appwrite/Platform/Tasks/ScheduleMessage.php index cd29a1dcf8..8db2efe2de 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleMessage.php +++ b/src/Appwrite/Platform/Tasks/ScheduleMessage.php @@ -133,7 +133,7 @@ class ScheduleMessage extends Action */ $enqueueMessages = function () use (&$schedules, $pools, $dbForConsole) { foreach ($schedules as $scheduleId => $schedule) { - \go(function () use ($schedules, $schedule, $pools, $dbForConsole) { + \go(function () use (&$schedules, $schedule, $pools, $dbForConsole) { $queue = $pools->get('queue')->pop(); $connection = $queue->getResource(); $queueForMessaging = new Messaging($connection); From b9892b857a6c075054dff65ad03c32986e34756c Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 11 Jan 2024 15:54:44 +1300 Subject: [PATCH 03/30] Make const int easier to read --- app/init.php | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/app/init.php b/app/init.php index 11cfef20ad..44bd3ac33c 100644 --- a/app/init.php +++ b/app/init.php @@ -88,17 +88,17 @@ const APP_MODE_DEFAULT = 'default'; const APP_MODE_ADMIN = 'admin'; const APP_PAGING_LIMIT = 12; const APP_LIMIT_COUNT = 5000; -const APP_LIMIT_USERS = 10000; +const APP_LIMIT_USERS = 10_000; const APP_LIMIT_USER_PASSWORD_HISTORY = 20; const APP_LIMIT_USER_SESSIONS_MAX = 100; const APP_LIMIT_USER_SESSIONS_DEFAULT = 10; -const APP_LIMIT_ANTIVIRUS = 20000000; //20MB -const APP_LIMIT_ENCRYPTION = 20000000; //20MB -const APP_LIMIT_COMPRESSION = 20000000; //20MB +const APP_LIMIT_ANTIVIRUS = 20_000_000; //20MB +const APP_LIMIT_ENCRYPTION = 20_000_000; //20MB +const APP_LIMIT_COMPRESSION = 20_000_000; //20MB const APP_LIMIT_ARRAY_PARAMS_SIZE = 100; // Default maximum of how many elements can there be in API parameter that expects array value const APP_LIMIT_ARRAY_ELEMENT_SIZE = 4096; // Default maximum length of element in array parameter represented by maximum URL length. const APP_LIMIT_SUBQUERY = 1000; -const APP_LIMIT_SUBSCRIBERS_SUBQUERY = 1000000; +const APP_LIMIT_SUBSCRIBERS_SUBQUERY = 1_000_000; const APP_LIMIT_WRITE_RATE_DEFAULT = 60; // Default maximum write rate per rate period const APP_LIMIT_WRITE_RATE_PERIOD_DEFAULT = 60; // Default maximum write rate period in seconds const APP_LIMIT_LIST_DEFAULT = 25; // Default maximum number of items to return in list API calls @@ -114,8 +114,8 @@ const APP_DATABASE_ATTRIBUTE_DATETIME = 'datetime'; const APP_DATABASE_ATTRIBUTE_URL = 'url'; const APP_DATABASE_ATTRIBUTE_INT_RANGE = 'intRange'; const APP_DATABASE_ATTRIBUTE_FLOAT_RANGE = 'floatRange'; -const APP_DATABASE_ATTRIBUTE_STRING_MAX_LENGTH = 1073741824; // 2^32 bits / 4 bits per char -const APP_DATABASE_TIMEOUT_MILLISECONDS = 15000; +const APP_DATABASE_ATTRIBUTE_STRING_MAX_LENGTH = 1_073_741_824; // 2^32 bits / 4 bits per char +const APP_DATABASE_TIMEOUT_MILLISECONDS = 15_000; const APP_STORAGE_UPLOADS = '/storage/uploads'; const APP_STORAGE_FUNCTIONS = '/storage/functions'; const APP_STORAGE_BUILDS = '/storage/builds'; From 6904285560e9ed7d87195663e109186e5913f78a Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 11 Jan 2024 15:55:08 +1300 Subject: [PATCH 04/30] Check provider enabled --- src/Appwrite/Platform/Workers/Messaging.php | 108 ++++++++++++-------- 1 file changed, 67 insertions(+), 41 deletions(-) diff --git a/src/Appwrite/Platform/Workers/Messaging.php b/src/Appwrite/Platform/Workers/Messaging.php index c4f117d37b..2d8770e29f 100644 --- a/src/Appwrite/Platform/Workers/Messaging.php +++ b/src/Appwrite/Platform/Workers/Messaging.php @@ -37,7 +37,7 @@ class Messaging extends Action { public static function getName(): string { - return "messaging"; + return 'messaging'; } /** @@ -67,10 +67,13 @@ class Messaging extends Action return; } - if (!\is_null($payload['message']) && !\is_null($payload['recipients'])) { - if ($payload['providerType'] === MESSAGE_TYPE_SMS) { - $this->processInternalSMSMessage(new Document($payload['message']), $payload['recipients']); - } + if ( + !\is_null($payload['message']) + && !\is_null($payload['recipients']) + && $payload['providerType'] === MESSAGE_TYPE_SMS + ) { + // Message was triggered internally + $this->processInternalSMSMessage(new Document($payload['message']), $payload['recipients']); } else { $message = $dbForProject->getDocument('messages', $payload['messageId']); @@ -80,84 +83,103 @@ class Messaging extends Action private function processMessage(Database $dbForProject, Document $message): void { - $topicsId = $message->getAttribute('topics', []); - $targetsId = $message->getAttribute('targets', []); - $usersId = $message->getAttribute('users', []); + $topicIds = $message->getAttribute('topics', []); + $targetIds = $message->getAttribute('targets', []); + $userIds = $message->getAttribute('users', []); /** - * @var Document[] $recipients + * @var array $recipients */ $recipients = []; - if (\count($topicsId) > 0) { - $topics = $dbForProject->find('topics', [Query::equal('$id', $topicsId)]); + if (\count($topicIds) > 0) { + $topics = $dbForProject->find('topics', [ + Query::equal('$id', $topicIds), + Query::limit(APP_LIMIT_SUBSCRIBERS_SUBQUERY) + ]); foreach ($topics as $topic) { - $targets = \array_filter($topic->getAttribute('targets'), fn(Document $target) => $target->getAttribute('providerType') === $message->getAttribute('providerType')); + $targets = \array_filter($topic->getAttribute('targets'), fn(Document $target) => + $target->getAttribute('providerType') === $message->getAttribute('providerType')); $recipients = \array_merge($recipients, $targets); } } - if (\count($usersId) > 0) { - $users = $dbForProject->find('users', [Query::equal('$id', $usersId)]); + if (\count($userIds) > 0) { + $users = $dbForProject->find('users', [ + Query::equal('$id', $userIds), + Query::limit(APP_LIMIT_SUBSCRIBERS_SUBQUERY) + ]); foreach ($users as $user) { - $targets = \array_filter($user->getAttribute('targets'), fn(Document $target) => $target->getAttribute('providerType') === $message->getAttribute('providerType')); + $targets = \array_filter($user->getAttribute('targets'), fn(Document $target) => + $target->getAttribute('providerType') === $message->getAttribute('providerType')); $recipients = \array_merge($recipients, $targets); } } - if (\count($targetsId) > 0) { - $targets = $dbForProject->find('targets', [Query::equal('$id', $targetsId)]); + if (\count($targetIds) > 0) { + $targets = $dbForProject->find('targets', [ + Query::equal('$id', $targetIds), + Query::limit(APP_LIMIT_SUBSCRIBERS_SUBQUERY) + ]); $recipients = \array_merge($recipients, $targets); } - $primaryProvider = $dbForProject->findOne('providers', [ + + $fallback = $dbForProject->findOne('providers', [ Query::equal('enabled', [true]), Query::equal('type', [$recipients[0]->getAttribute('providerType')]), ]); /** - * @var array> $identifiersByProviderId + * @var array> $identifiers */ - $identifiersByProviderId = []; + $identifiers = []; /** * @var Document[] $providers */ $providers = [ - $primaryProvider->getId() => $primaryProvider + $fallback->getId() => $fallback ]; + foreach ($recipients as $recipient) { $providerId = $recipient->getAttribute('providerId'); - if (!$providerId && $primaryProvider instanceof Document && !$primaryProvider->isEmpty()) { - $providerId = $primaryProvider->getId(); + if ( + !$providerId + && $fallback instanceof Document + && !$fallback->isEmpty() + && $fallback->getAttribute('enabled') + ) { + $providerId = $fallback->getId(); } if ($providerId) { - if (!isset($identifiersByProviderId[$providerId])) { - $identifiersByProviderId[$providerId] = []; + if (!\array_key_exists($providerId, $identifiers)) { + $identifiers[$providerId] = []; } - $identifiersByProviderId[$providerId][] = $recipient->getAttribute('identifier'); + $identifiers[$providerId][] = $recipient->getAttribute('identifier'); } } /** - * @var array[] $results + * @var array $results */ - $results = batch(\array_map(function ($providerId) use ($identifiersByProviderId, $providers, $primaryProvider, $message, $dbForProject) { - return function () use ($providerId, $identifiersByProviderId, $providers, $primaryProvider, $message, $dbForProject) { + $results = batch(\array_map(function ($providerId) use ($identifiers, $providers, $fallback, $message, $dbForProject) { + return function () use ($providerId, $identifiers, $providers, $fallback, $message, $dbForProject) { if (\array_key_exists($providerId, $providers)) { $provider = $providers[$providerId]; } else { - $provider = $dbForProject->getDocument('providers', $providerId, [Query::equal('enabled', [true])]); + $provider = $dbForProject->getDocument('providers', $providerId); - if ($provider->isEmpty()) { - $provider = $primaryProvider; + if ($provider->isEmpty() || !$provider->getAttribute('enabled')) { + $provider = $fallback; } else { $providers[$providerId] = $provider; } } - $identifiers = $identifiersByProviderId[$providerId]; + $identifiers = $identifiers[$providerId]; + $adapter = match ($provider->getAttribute('type')) { MESSAGE_TYPE_SMS => $this->sms($provider), MESSAGE_TYPE_PUSH => $this->push($provider), @@ -196,7 +218,10 @@ class Messaging extends Action // Deleting push targets when token has expired. if ($detail['error'] === 'Expired device token.') { - $target = $dbForProject->findOne('targets', [Query::equal('identifier', [$detail['recipient']])]); + $target = $dbForProject->findOne('targets', [ + Query::equal('identifier', [$detail['recipient']]) + ]); + if ($target instanceof Document && !$target->isEmpty()) { $dbForProject->deleteDocument('targets', $target->getId()); } @@ -206,6 +231,7 @@ class Messaging extends Action $deliveryErrors[] = 'Failed sending to targets ' . $batchIndex + 1 . '-' . \count($batch) . ' with error: ' . $e->getMessage(); } finally { $batchIndex++; + return [ 'deliveredTotal' => $deliveredTotal, 'deliveryErrors' => $deliveryErrors, @@ -214,7 +240,7 @@ class Messaging extends Action }; }, $batches)); }; - }, \array_keys($identifiersByProviderId))); + }, \array_keys($identifiers))); $results = array_merge(...$results); @@ -229,6 +255,7 @@ class Messaging extends Action $message->setAttribute('deliveryErrors', $deliveryErrors); if (\count($message->getAttribute('deliveryErrors')) > 0) { + // TODO: Does this make sense? Only some of the messages might have failed, but we mark the whole message as failed. $message->setAttribute('status', 'failed'); } else { $message->setAttribute('status', 'sent'); @@ -249,15 +276,14 @@ class Messaging extends Action private function processInternalSMSMessage(Document $message, array $recipients): void { if (empty(App::getEnv('_APP_SMS_PROVIDER')) || empty(App::getEnv('_APP_SMS_FROM'))) { - Console::info('Skipped SMS processing. No Phone configuration has been set.'); + Console::info('Skipped SMS processing. Missing "_APP_SMS_PROVIDER" or "_APP_SMS_FROM" environment variables.'); return; } - $smsDSN = new DSN(App::getEnv('_APP_SMS_PROVIDER')); - $host = $smsDSN->getHost(); - $password = $smsDSN->getPassword(); - $user = $smsDSN->getUser(); - + $dsn = new DSN(App::getEnv('_APP_SMS_PROVIDER')); + $host = $dsn->getHost(); + $password = $dsn->getPassword(); + $user = $dsn->getUser(); $from = App::getEnv('_APP_SMS_FROM'); $provider = new Document([ From 9063b4e77f2382a863e1cca52de24267481874c3 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 11 Jan 2024 15:58:24 +1300 Subject: [PATCH 05/30] Fix controller scheduling, ensuring created when message updated and none exists --- app/config/errors.php | 13 +- app/controllers/api/messaging.php | 188 ++++++++++--------- tests/e2e/Services/GraphQL/Base.php | 16 +- tests/e2e/Services/GraphQL/MessagingTest.php | 10 +- 4 files changed, 118 insertions(+), 109 deletions(-) diff --git a/app/config/errors.php b/app/config/errors.php index 2e35bfb881..bc2c436313 100644 --- a/app/config/errors.php +++ b/app/config/errors.php @@ -807,7 +807,7 @@ return [ ], Exception::PROVIDER_INCORRECT_TYPE => [ 'name' => Exception::PROVIDER_INCORRECT_TYPE, - 'description' => 'Provider with the requested ID is of incorrect type: ', + 'description' => 'Provider with the requested ID is of the incorrect type.', 'code' => 400, ], @@ -858,18 +858,23 @@ return [ ], Exception::MESSAGE_TARGET_NOT_EMAIL => [ 'name' => Exception::MESSAGE_TARGET_NOT_EMAIL, - 'description' => 'Message with the target ID is not an email target:', + 'description' => 'Message with the target ID is not an email target.', 'code' => 400, ], Exception::MESSAGE_TARGET_NOT_SMS => [ 'name' => Exception::MESSAGE_TARGET_NOT_SMS, - 'description' => 'Message with the target ID is not an SMS target:', + 'description' => 'Message with the target ID is not an SMS target.', 'code' => 400, ], Exception::MESSAGE_TARGET_NOT_PUSH => [ 'name' => Exception::MESSAGE_TARGET_NOT_PUSH, - 'description' => 'Message with the target ID is not a push target:', + 'description' => 'Message with the target ID is not a push target.', 'code' => 400, ], + Exception::SCHEDULE_NOT_FOUND => [ + 'name' => Exception::SCHEDULE_NOT_FOUND, + 'description' => 'Schedule with the requested ID could not be found.', + 'code' => 404, + ], ]; diff --git a/app/controllers/api/messaging.php b/app/controllers/api/messaging.php index 1c85bc1839..6f52ed2fe7 100644 --- a/app/controllers/api/messaging.php +++ b/app/controllers/api/messaging.php @@ -602,7 +602,7 @@ App::post('/v1/messaging/providers/fcm') ->label('scope', 'providers.write') ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY]) ->label('sdk.namespace', 'messaging') - ->label('sdk.method', 'createFcmProvider') + ->label('sdk.method', 'createFCMProvider') ->label('sdk.description', '/docs/references/messaging/create-fcm-provider.md') ->label('sdk.response.code', Response::STATUS_CODE_CREATED) ->label('sdk.response.type', Response::CONTENT_TYPE_JSON) @@ -661,7 +661,7 @@ App::post('/v1/messaging/providers/apns') ->label('scope', 'providers.write') ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY]) ->label('sdk.namespace', 'messaging') - ->label('sdk.method', 'createApnsProvider') + ->label('sdk.method', 'createAPNSProvider') ->label('sdk.description', '/docs/references/messaging/create-apns-provider.md') ->label('sdk.response.code', Response::STATUS_CODE_CREATED) ->label('sdk.response.type', Response::CONTENT_TYPE_JSON) @@ -1491,8 +1491,9 @@ App::patch('/v1/messaging/providers/fcm/:providerId') ->label('audits.resource', 'provider/{response.$id}') ->label('event', 'providers.[providerId].update') ->label('scope', 'providers.write') + ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY]) ->label('sdk.namespace', 'messaging') - ->label('sdk.method', 'updateFcmProvider') + ->label('sdk.method', 'updateFCMProvider') ->label('sdk.description', '/docs/references/messaging/update-fcm-provider.md') ->label('sdk.response.code', Response::STATUS_CODE_OK) ->label('sdk.response.type', Response::CONTENT_TYPE_JSON) @@ -1553,7 +1554,7 @@ App::patch('/v1/messaging/providers/apns/:providerId') ->label('scope', 'providers.write') ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY]) ->label('sdk.namespace', 'messaging') - ->label('sdk.method', 'updateApnsProvider') + ->label('sdk.method', 'updateAPNSProvider') ->label('sdk.description', '/docs/references/messaging/update-apns-provider.md') ->label('sdk.response.code', Response::STATUS_CODE_OK) ->label('sdk.response.type', Response::CONTENT_TYPE_JSON) @@ -2268,7 +2269,7 @@ App::post('/v1/messaging/messages/email') ->label('scope', 'messages.write') ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY]) ->label('sdk.namespace', 'messaging') - ->label('sdk.method', 'createEmailMessage') + ->label('sdk.method', 'createEmail') ->label('sdk.description', '/docs/references/messaging/create-email.md') ->label('sdk.response.code', Response::STATUS_CODE_CREATED) ->label('sdk.response.type', Response::CONTENT_TYPE_JSON) @@ -2282,7 +2283,7 @@ App::post('/v1/messaging/messages/email') ->param('cc', [], new ArrayList(new UID()), 'Array of target IDs to be added as CC.', true) ->param('bcc', [], new ArrayList(new UID()), 'Array of target IDs to be added as BCC.', true) ->param('description', '', new Text(256), 'Description for message.', true) - ->param('status', 'processing', new WhiteList(['draft', 'canceled', 'processing']), 'Message Status. Value must be either draft or cancelled or processing.', true) + ->param('status', 'processing', new WhiteList(['draft', 'processing']), 'Message Status. Value must be either draft or cancelled or processing.', true) ->param('html', false, new Boolean(), 'Is content of type HTML', true) ->param('scheduledAt', null, new DatetimeValidator(requireDateInFuture: true), 'Scheduled delivery time for message in [ISO 8601](https://www.iso.org/iso-8601-date-and-time-format.html) format. DateTime value must be in future.', true) ->inject('queueForEvents') @@ -2338,11 +2339,11 @@ App::post('/v1/messaging/messages/email') 'status' => $status, ])); - if ($status === 'processing' && $scheduledAt === null) { + if ($status === 'processing' && \is_null($scheduledAt)) { $queueForMessaging ->setMessageId($message->getId()) ->trigger(); - } elseif ($scheduledAt !== null) { + } elseif (!\is_null($scheduledAt)) { $schedule = $dbForConsole->createDocument('schedules', new Document([ 'region' => App::getEnv('_APP_REGION', 'default'), 'resourceType' => 'message', @@ -2351,11 +2352,16 @@ App::post('/v1/messaging/messages/email') 'resourceUpdatedAt' => DateTime::now(), 'projectId' => $project->getId(), 'schedule' => $message->getAttribute('scheduledAt'), - 'active' => $status === 'processing' ? true : false, + 'active' => $status === 'processing', ])); $message->setAttribute('scheduleId', $schedule->getId()); - $dbForProject->updateDocument('messages', $message->getId(), $message); + + $dbForProject->updateDocument( + 'messages', + $message->getId(), + $message + ); } $queueForEvents @@ -2375,7 +2381,7 @@ App::post('/v1/messaging/messages/sms') ->label('scope', 'messages.write') ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY]) ->label('sdk.namespace', 'messaging') - ->label('sdk.method', 'createSMSMessage') + ->label('sdk.method', 'createSMS') ->label('sdk.description', '/docs/references/messaging/create-sms.md') ->label('sdk.response.code', Response::STATUS_CODE_CREATED) ->label('sdk.response.type', Response::CONTENT_TYPE_JSON) @@ -2445,7 +2451,7 @@ App::post('/v1/messaging/messages/sms') 'resourceUpdatedAt' => DateTime::now(), 'projectId' => $project->getId(), 'schedule' => $message->getAttribute('scheduledAt'), - 'active' => $status === 'processing' ? true : false, + 'active' => $status === 'processing', ])); $message->setAttribute('scheduleId', $schedule->getId()); @@ -2469,7 +2475,7 @@ App::post('/v1/messaging/messages/push') ->label('scope', 'messages.write') ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY]) ->label('sdk.namespace', 'messaging') - ->label('sdk.method', 'createPushMessage') + ->label('sdk.method', 'createPush') ->label('sdk.description', '/docs/references/messaging/create-push-notification.md') ->label('sdk.response.code', Response::STATUS_CODE_CREATED) ->label('sdk.response.type', Response::CONTENT_TYPE_JSON) @@ -2556,7 +2562,7 @@ App::post('/v1/messaging/messages/push') 'resourceUpdatedAt' => DateTime::now(), 'projectId' => $project->getId(), 'schedule' => $message->getAttribute('scheduledAt'), - 'active' => $status === 'processing' ? true : false, + 'active' => $status === 'processing', ])); $message->setAttribute('scheduleId', $schedule->getId()); @@ -2777,30 +2783,12 @@ App::patch('/v1/messaging/messages/email/:messageId') $message->setAttribute('users', $users); } - if (!\is_null($targets) || !\is_null($cc) || !\is_null($bcc)) { - $mergedTargets = \array_merge(...\array_filter([$targets, $cc, $bcc])); - - $foundTargets = $dbForProject->find('targets', [ - Query::equal('$id', $mergedTargets), - Query::equal('providerType', [MESSAGE_TYPE_EMAIL]), - Query::limit(\count($mergedTargets)), - ]); - if (\count($foundTargets) !== \count($mergedTargets)) { - throw new Exception(Exception::MESSAGE_TARGET_NOT_EMAIL); - } - foreach ($foundTargets as $target) { - if ($target->isEmpty()) { - throw new Exception(Exception::USER_TARGET_NOT_FOUND); - } - } - } - - $data = $message->getAttribute('data'); - if (!\is_null($targets)) { $message->setAttribute('targets', $targets); } + $data = $message->getAttribute('data'); + if (!\is_null($subject)) { $data['subject'] = $subject; } @@ -2831,20 +2819,36 @@ App::patch('/v1/messaging/messages/email/:messageId') $message->setAttribute('status', $status); } - if (!is_null($scheduledAt)) { - $message->setAttribute('scheduledAt', $scheduledAt); + if (!\is_null($scheduledAt)) { + if (\is_null($message->getAttribute(('scheduleId')))) { + $schedule = $dbForConsole->createDocument('schedules', new Document([ + 'region' => App::getEnv('_APP_REGION', 'default'), + 'resourceType' => 'message', + 'resourceId' => $message->getId(), + 'resourceInternalId' => $message->getInternalId(), + 'resourceUpdatedAt' => DateTime::now(), + 'projectId' => $project->getId(), + 'schedule' => $scheduledAt, + 'active' => $status === 'processing', + ])); - $schedule = $dbForConsole->getDocument('schedules', $message->getAttribute('scheduleId')); + $message->setAttribute('scheduleId', $schedule->getId()); + } else { + $schedule = $dbForConsole->getDocument('schedules', $message->getAttribute('scheduleId')); - $schedule - ->setAttribute('resourceUpdatedAt', DateTime::now()) - ->setAttribute('schedule', $message->getAttribute('schedule')); + if ($schedule->isEmpty()) { + throw new Exception(Exception::SCHEDULE_NOT_FOUND); + } - if ($message->getAttribute('status') === 'processing') { - $schedule->setAttribute('active', true); + $schedule + ->setAttribute('resourceUpdatedAt', DateTime::now()) + ->setAttribute('schedule', $scheduledAt) + ->setAttribute('active', $status === 'processing'); + + $dbForConsole->updateDocument('schedules', $schedule->getId(), $schedule); } - $dbForConsole->updateDocument('schedules', $schedule->getId(), $schedule); + $message->setAttribute('scheduleId', $schedule->getId()); } $message = $dbForProject->updateDocument('messages', $message->getId(), $message); @@ -2914,22 +2918,6 @@ App::patch('/v1/messaging/messages/sms/:messageId') } if (!\is_null($targets)) { - $foundTargets = $dbForProject->find('targets', [ - Query::equal('$id', $targets), - Query::equal('providerType', [MESSAGE_TYPE_SMS]), - Query::limit(\count($targets)), - ]); - - if (\count($foundTargets) !== \count($targets)) { - throw new Exception(Exception::MESSAGE_TARGET_NOT_SMS); - } - - foreach ($foundTargets as $target) { - if ($target->isEmpty()) { - throw new Exception(Exception::USER_TARGET_NOT_FOUND); - } - } - $message->setAttribute('targets', $targets); } @@ -2949,20 +2937,36 @@ App::patch('/v1/messaging/messages/sms/:messageId') $message->setAttribute('description', $description); } - if (!is_null($scheduledAt)) { - $message->setAttribute('scheduledAt', $scheduledAt); + if (!\is_null($scheduledAt)) { + if (\is_null($message->getAttribute(('scheduleId')))) { + $schedule = $dbForConsole->createDocument('schedules', new Document([ + 'region' => App::getEnv('_APP_REGION', 'default'), + 'resourceType' => 'message', + 'resourceId' => $message->getId(), + 'resourceInternalId' => $message->getInternalId(), + 'resourceUpdatedAt' => DateTime::now(), + 'projectId' => $project->getId(), + 'schedule' => $scheduledAt, + 'active' => $status === 'processing', + ])); - $schedule = $dbForConsole->getDocument('schedules', $message->getAttribute('scheduleId')); + $message->setAttribute('scheduleId', $schedule->getId()); + } else { + $schedule = $dbForConsole->getDocument('schedules', $message->getAttribute('scheduleId')); - $schedule - ->setAttribute('resourceUpdatedAt', DateTime::now()) - ->setAttribute('schedule', $message->getAttribute('schedule')); + if ($schedule->isEmpty()) { + throw new Exception(Exception::SCHEDULE_NOT_FOUND); + } - if ($message->getAttribute('status') === 'processing') { - $schedule->setAttribute('active', true); + $schedule + ->setAttribute('resourceUpdatedAt', DateTime::now()) + ->setAttribute('schedule', $scheduledAt) + ->setAttribute('active', $status === 'processing'); + + $dbForConsole->updateDocument('schedules', $schedule->getId(), $schedule); } - $dbForConsole->updateDocument('schedules', $schedule->getId(), $schedule); + $message->setAttribute('scheduleId', $schedule->getId()); } $message = $dbForProject->updateDocument('messages', $message->getId(), $message); @@ -3040,22 +3044,6 @@ App::patch('/v1/messaging/messages/push/:messageId') } if (!\is_null($targets)) { - $foundTargets = $dbForProject->find('targets', [ - Query::equal('$id', $targets), - Query::equal('providerType', [MESSAGE_TYPE_PUSH]), - Query::limit(\count($targets)), - ]); - - if (\count($foundTargets) !== \count($targets)) { - throw new Exception(Exception::MESSAGE_TARGET_NOT_PUSH); - } - - foreach ($foundTargets as $target) { - if ($target->isEmpty()) { - throw new Exception(Exception::USER_TARGET_NOT_FOUND); - } - } - $message->setAttribute('targets', $targets); } @@ -3108,19 +3096,35 @@ App::patch('/v1/messaging/messages/push/:messageId') } if (!\is_null($scheduledAt)) { - $message->setAttribute('scheduledAt', $scheduledAt); + if (\is_null($message->getAttribute(('scheduleId')))) { + $schedule = $dbForConsole->createDocument('schedules', new Document([ + 'region' => App::getEnv('_APP_REGION', 'default'), + 'resourceType' => 'message', + 'resourceId' => $message->getId(), + 'resourceInternalId' => $message->getInternalId(), + 'resourceUpdatedAt' => DateTime::now(), + 'projectId' => $project->getId(), + 'schedule' => $scheduledAt, + 'active' => $status === 'processing', + ])); - $schedule = $dbForConsole->getDocument('schedules', $message->getAttribute('scheduleId')); + $message->setAttribute('scheduleId', $schedule->getId()); + } else { + $schedule = $dbForConsole->getDocument('schedules', $message->getAttribute('scheduleId')); - $schedule - ->setAttribute('resourceUpdatedAt', DateTime::now()) - ->setAttribute('schedule', $message->getAttribute('schedule')); + if ($schedule->isEmpty()) { + throw new Exception(Exception::SCHEDULE_NOT_FOUND); + } - if ($message->getAttribute('status') === 'processing') { - $schedule->setAttribute('active', true); + $schedule + ->setAttribute('resourceUpdatedAt', DateTime::now()) + ->setAttribute('schedule', $scheduledAt) + ->setAttribute('active', $status === 'processing'); + + $dbForConsole->updateDocument('schedules', $schedule->getId(), $schedule); } - $dbForConsole->updateDocument('schedules', $schedule->getId(), $schedule); + $message->setAttribute('scheduleId', $schedule->getId()); } $message = $dbForProject->updateDocument('messages', $message->getId(), $message); diff --git a/tests/e2e/Services/GraphQL/Base.php b/tests/e2e/Services/GraphQL/Base.php index 2c630e3f75..c89eb88672 100644 --- a/tests/e2e/Services/GraphQL/Base.php +++ b/tests/e2e/Services/GraphQL/Base.php @@ -1860,8 +1860,8 @@ trait Base } }'; case self::$CREATE_FCM_PROVIDER: - return 'mutation createFcmProvider($providerId: String!, $name: String!, $serviceAccountJSON: Json) { - messagingCreateFcmProvider(providerId: $providerId, name: $name, serviceAccountJSON: $serviceAccountJSON) { + return 'mutation createFCMProvider($providerId: String!, $name: String!, $serviceAccountJSON: Json) { + messagingCreateFCMProvider(providerId: $providerId, name: $name, serviceAccountJSON: $serviceAccountJSON) { _id name provider @@ -1870,8 +1870,8 @@ trait Base } }'; case self::$CREATE_APNS_PROVIDER: - return 'mutation createApnsProvider($providerId: String!, $name: String!, $authKey: String!, $authKeyId: String!, $teamId: String!, $bundleId: String!, $endpoint: String!) { - messagingCreateApnsProvider(providerId: $providerId, name: $name, authKey: $authKey, authKeyId: $authKeyId, teamId: $teamId, bundleId: $bundleId, endpoint: $endpoint) { + return 'mutation createAPNSProvider($providerId: String!, $name: String!, $authKey: String!, $authKeyId: String!, $teamId: String!, $bundleId: String!, $endpoint: String!) { + messagingCreateAPNSProvider(providerId: $providerId, name: $name, authKey: $authKey, authKeyId: $authKeyId, teamId: $teamId, bundleId: $bundleId, endpoint: $endpoint) { _id name provider @@ -1974,8 +1974,8 @@ trait Base } }'; case self::$UPDATE_FCM_PROVIDER: - return 'mutation updateFcmProvider($providerId: String!, $name: String!, $serviceAccountJSON: Json) { - messagingUpdateFcmProvider(providerId: $providerId, name: $name, serviceAccountJSON: $serviceAccountJSON) { + return 'mutation updateFCMProvider($providerId: String!, $name: String!, $serviceAccountJSON: Json) { + messagingUpdateFCMProvider(providerId: $providerId, name: $name, serviceAccountJSON: $serviceAccountJSON) { _id name provider @@ -1984,8 +1984,8 @@ trait Base } }'; case self::$UPDATE_APNS_PROVIDER: - return 'mutation updateApnsProvider($providerId: String!, $name: String!, $authKey: String!, $authKeyId: String!, $teamId: String!, $bundleId: String!, $endpoint: String!) { - messagingUpdateApnsProvider(providerId: $providerId, name: $name, authKey: $authKey, authKeyId: $authKeyId, teamId: $teamId, bundleId: $bundleId, endpoint: $endpoint) { + return 'mutation updateAPNSProvider($providerId: String!, $name: String!, $authKey: String!, $authKeyId: String!, $teamId: String!, $bundleId: String!, $endpoint: String!) { + messagingUpdateAPNSProvider(providerId: $providerId, name: $name, authKey: $authKey, authKeyId: $authKeyId, teamId: $teamId, bundleId: $bundleId, endpoint: $endpoint) { _id name provider diff --git a/tests/e2e/Services/GraphQL/MessagingTest.php b/tests/e2e/Services/GraphQL/MessagingTest.php index 1828411483..2ca35e1d4e 100644 --- a/tests/e2e/Services/GraphQL/MessagingTest.php +++ b/tests/e2e/Services/GraphQL/MessagingTest.php @@ -70,7 +70,7 @@ class MessagingTest extends Scope 'apiSecret' => 'my-apisecret', 'from' => '+123456789', ], - 'Fcm' => [ + 'FCM' => [ 'providerId' => ID::unique(), 'name' => 'FCM1', 'serviceAccountJSON' => [ @@ -80,7 +80,7 @@ class MessagingTest extends Scope "private_key" => "test-private-key", ] ], - 'Apns' => [ + 'APNS' => [ 'providerId' => ID::unique(), 'name' => 'APNS1', 'authKey' => 'my-authkey', @@ -160,7 +160,7 @@ class MessagingTest extends Scope 'apiKey' => 'my-apikey', 'apiSecret' => 'my-apisecret', ], - 'Fcm' => [ + 'FCM' => [ 'providerId' => $providers[7]['_id'], 'name' => 'FCM2', 'serviceAccountJSON' => [ @@ -170,7 +170,7 @@ class MessagingTest extends Scope 'private_key' => "test-private-key", ] ], - 'Apns' => [ + 'APNS' => [ 'providerId' => $providers[8]['_id'], 'name' => 'APNS2', 'authKey' => 'my-authkey', @@ -1000,7 +1000,7 @@ class MessagingTest extends Scope $this->assertEquals(200, $provider['headers']['status-code']); - $providerId = $provider['body']['data']['messagingCreateFcmProvider']['_id']; + $providerId = $provider['body']['data']['messagingCreateFCMProvider']['_id']; $query = $this->getQuery(self::$CREATE_TOPIC); $graphQLPayload = [ From a40c6fce6429334d5c752c2ab275cd1f0e8348b5 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 11 Jan 2024 16:00:48 +1300 Subject: [PATCH 06/30] Add resource collection to schedules --- app/config/collections.php | 11 +++++++++++ app/controllers/api/functions.php | 1 + app/controllers/api/messaging.php | 6 ++++++ 3 files changed, 18 insertions(+) diff --git a/app/config/collections.php b/app/config/collections.php index a2af46293f..c82b6b671e 100644 --- a/app/config/collections.php +++ b/app/config/collections.php @@ -4142,6 +4142,17 @@ $consoleCollections = array_merge([ 'array' => false, 'filters' => [], ], + [ + '$id' => ID::custom('resourceCollection'), + 'type' => Database::VAR_STRING, + 'format' => '', + 'size' => Database::LENGTH_KEY, + 'signed' => true, + 'required' => true, + 'default' => null, + 'array' => false, + 'filters' => [], + ], [ '$id' => ID::custom('resourceInternalId'), 'type' => Database::VAR_STRING, diff --git a/app/controllers/api/functions.php b/app/controllers/api/functions.php index cbdbd3a1cb..154f194375 100644 --- a/app/controllers/api/functions.php +++ b/app/controllers/api/functions.php @@ -228,6 +228,7 @@ App::post('/v1/functions') fn () => $dbForConsole->createDocument('schedules', new Document([ 'region' => App::getEnv('_APP_REGION', 'default'), // Todo replace with projects region 'resourceType' => 'function', + 'resourceCollection' => 'functions', 'resourceId' => $function->getId(), 'resourceInternalId' => $function->getInternalId(), 'resourceUpdatedAt' => DateTime::now(), diff --git a/app/controllers/api/messaging.php b/app/controllers/api/messaging.php index 6f52ed2fe7..2471ab8b26 100644 --- a/app/controllers/api/messaging.php +++ b/app/controllers/api/messaging.php @@ -2347,6 +2347,7 @@ App::post('/v1/messaging/messages/email') $schedule = $dbForConsole->createDocument('schedules', new Document([ 'region' => App::getEnv('_APP_REGION', 'default'), 'resourceType' => 'message', + 'resourceCollection' => 'messages', 'resourceId' => $message->getId(), 'resourceInternalId' => $message->getInternalId(), 'resourceUpdatedAt' => DateTime::now(), @@ -2446,6 +2447,7 @@ App::post('/v1/messaging/messages/sms') $schedule = $dbForConsole->createDocument('schedules', new Document([ 'region' => App::getEnv('_APP_REGION', 'default'), 'resourceType' => 'message', + 'resourceCollection' => 'messages', 'resourceId' => $message->getId(), 'resourceInternalId' => $message->getInternalId(), 'resourceUpdatedAt' => DateTime::now(), @@ -2557,6 +2559,7 @@ App::post('/v1/messaging/messages/push') $schedule = $dbForConsole->createDocument('schedules', new Document([ 'region' => App::getEnv('_APP_REGION', 'default'), 'resourceType' => 'message', + 'resourceCollection' => 'messages', 'resourceId' => $message->getId(), 'resourceInternalId' => $message->getInternalId(), 'resourceUpdatedAt' => DateTime::now(), @@ -2824,6 +2827,7 @@ App::patch('/v1/messaging/messages/email/:messageId') $schedule = $dbForConsole->createDocument('schedules', new Document([ 'region' => App::getEnv('_APP_REGION', 'default'), 'resourceType' => 'message', + 'resourceCollection' => 'messages', 'resourceId' => $message->getId(), 'resourceInternalId' => $message->getInternalId(), 'resourceUpdatedAt' => DateTime::now(), @@ -2942,6 +2946,7 @@ App::patch('/v1/messaging/messages/sms/:messageId') $schedule = $dbForConsole->createDocument('schedules', new Document([ 'region' => App::getEnv('_APP_REGION', 'default'), 'resourceType' => 'message', + 'resourceCollection' => 'messages', 'resourceId' => $message->getId(), 'resourceInternalId' => $message->getInternalId(), 'resourceUpdatedAt' => DateTime::now(), @@ -3100,6 +3105,7 @@ App::patch('/v1/messaging/messages/push/:messageId') $schedule = $dbForConsole->createDocument('schedules', new Document([ 'region' => App::getEnv('_APP_REGION', 'default'), 'resourceType' => 'message', + 'resourceCollection' => 'messages', 'resourceId' => $message->getId(), 'resourceInternalId' => $message->getInternalId(), 'resourceUpdatedAt' => DateTime::now(), From 81f1eb35060c08d4a72bbfdd67d87321f0ec4aea Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 11 Jan 2024 16:06:59 +1300 Subject: [PATCH 07/30] Abstract scheduling base --- src/Appwrite/Extend/Exception.php | 3 + src/Appwrite/Platform/Tasks/Schedule.php | 244 ------------------ src/Appwrite/Platform/Tasks/ScheduleBase.php | 187 ++++++++++++++ .../Platform/Tasks/ScheduleFunctions.php | 101 ++++++++ .../Platform/Tasks/ScheduleMessage.php | 162 ------------ .../Platform/Tasks/ScheduleMessages.php | 57 ++++ 6 files changed, 348 insertions(+), 406 deletions(-) delete mode 100644 src/Appwrite/Platform/Tasks/Schedule.php create mode 100644 src/Appwrite/Platform/Tasks/ScheduleBase.php create mode 100644 src/Appwrite/Platform/Tasks/ScheduleFunctions.php delete mode 100644 src/Appwrite/Platform/Tasks/ScheduleMessage.php create mode 100644 src/Appwrite/Platform/Tasks/ScheduleMessages.php diff --git a/src/Appwrite/Extend/Exception.php b/src/Appwrite/Extend/Exception.php index ea63423c05..b1d654c400 100644 --- a/src/Appwrite/Extend/Exception.php +++ b/src/Appwrite/Extend/Exception.php @@ -263,6 +263,9 @@ class Exception extends \Exception public const MESSAGE_TARGET_NOT_SMS = 'message_target_not_sms'; public const MESSAGE_TARGET_NOT_PUSH = 'message_target_not_push'; + /** Schedules */ + public const SCHEDULE_NOT_FOUND = 'schedule_not_found'; + protected string $type = ''; protected array $errors = []; diff --git a/src/Appwrite/Platform/Tasks/Schedule.php b/src/Appwrite/Platform/Tasks/Schedule.php deleted file mode 100644 index a136ee62b1..0000000000 --- a/src/Appwrite/Platform/Tasks/Schedule.php +++ /dev/null @@ -1,244 +0,0 @@ -desc('Execute functions scheduled in Appwrite') - ->inject('pools') - ->inject('dbForConsole') - ->inject('getProjectDB') - ->callback(fn (Group $pools, Database $dbForConsole, callable $getProjectDB) => $this->action($pools, $dbForConsole, $getProjectDB)); - } - - /** - * 1. Load all documents from 'schedules' collection to create local copy - * 2. Create timer that sync all changes from 'schedules' collection to local copy. Only reading changes thanks to 'resourceUpdatedAt' attribute - * 3. Create timer that prepares coroutines for soon-to-execute schedules. When it's ready, coroutime sleeps until exact time before sending request to worker. - */ - public function action(Group $pools, Database $dbForConsole, callable $getProjectDB): void - { - Console::title('Scheduler V1'); - Console::success(APP_NAME . ' Scheduler v1 has started'); - - /** - * Extract only nessessary attributes to lower memory used. - * - * @var Document $schedule - * @return array - */ - $getSchedule = function (Document $schedule) use ($dbForConsole, $getProjectDB): array { - $project = $dbForConsole->getDocument('projects', $schedule->getAttribute('projectId')); - - $function = $getProjectDB($project)->getDocument('functions', $schedule->getAttribute('resourceId')); - - return [ - 'resourceId' => $schedule->getAttribute('resourceId'), - 'schedule' => $schedule->getAttribute('schedule'), - 'resourceUpdatedAt' => $schedule->getAttribute('resourceUpdatedAt'), - 'project' => $project, // TODO: @Meldiron Send only ID to worker to reduce memory usage here - 'function' => $function, // TODO: @Meldiron Send only ID to worker to reduce memory usage here - ]; - }; - - $schedules = []; // Local copy of 'schedules' collection - $lastSyncUpdate = DateTime::now(); - - $limit = 10000; - $sum = $limit; - $total = 0; - $loadStart = \microtime(true); - $latestDocument = null; - - while ($sum === $limit) { - $paginationQueries = [Query::limit($limit)]; - if ($latestDocument !== null) { - $paginationQueries[] = Query::cursorAfter($latestDocument); - } - $results = $dbForConsole->find('schedules', \array_merge($paginationQueries, [ - Query::equal('region', [App::getEnv('_APP_REGION', 'default')]), - Query::equal('resourceType', ['function']), - Query::equal('active', [true]), - ])); - - $sum = count($results); - $total = $total + $sum; - foreach ($results as $document) { - try { - $schedules[$document['resourceId']] = $getSchedule($document); - } catch (\Throwable $th) { - Console::error("Failed to load schedule for project {$document['projectId']} and function {$document['resourceId']}"); - Console::error($th->getMessage()); - } - } - - $latestDocument = !empty(array_key_last($results)) ? $results[array_key_last($results)] : null; - } - - $pools->reclaim(); - - Console::success("{$total} functions were loaded in " . (microtime(true) - $loadStart) . " seconds"); - - Console::success("Starting timers at " . DateTime::now()); - - run( - function () use ($dbForConsole, &$schedules, &$lastSyncUpdate, $getSchedule, $pools) { - /** - * The timer synchronize $schedules copy with database collection. - */ - Timer::tick(self::FUNCTION_UPDATE_TIMER * 1000, function () use ($dbForConsole, &$schedules, &$lastSyncUpdate, $getSchedule, $pools) { - $time = DateTime::now(); - $timerStart = \microtime(true); - - $limit = 1000; - $sum = $limit; - $total = 0; - $latestDocument = null; - - Console::log("Sync tick: Running at $time"); - - while ($sum === $limit) { - $paginationQueries = [Query::limit($limit)]; - if ($latestDocument !== null) { - $paginationQueries[] = Query::cursorAfter($latestDocument); - } - $results = $dbForConsole->find('schedules', \array_merge($paginationQueries, [ - Query::equal('region', [App::getEnv('_APP_REGION', 'default')]), - Query::equal('resourceType', ['function']), - Query::greaterThanEqual('resourceUpdatedAt', $lastSyncUpdate), - ])); - - $sum = count($results); - $total = $total + $sum; - foreach ($results as $document) { - $localDocument = $schedules[$document['resourceId']] ?? null; - - $org = $localDocument !== null ? strtotime($localDocument['resourceUpdatedAt']) : null; - $new = strtotime($document['resourceUpdatedAt']); - - if ($document['active'] === false) { - Console::info("Removing: {$document['resourceId']}"); - unset($schedules[$document['resourceId']]); - } elseif ($new !== $org) { - Console::info("Updating: {$document['resourceId']}"); - $schedules[$document['resourceId']] = $getSchedule($document); - } - } - $latestDocument = !empty(array_key_last($results)) ? $results[array_key_last($results)] : null; - } - - $lastSyncUpdate = $time; - $timerEnd = \microtime(true); - - $pools->reclaim(); - - Console::log("Sync tick: {$total} schedules were updated in " . ($timerEnd - $timerStart) . " seconds"); - }); - - /** - * The timer to prepare soon-to-execute schedules. - */ - $lastEnqueueUpdate = null; - $enqueueFunctions = function () use (&$schedules, $lastEnqueueUpdate, $pools) { - $timerStart = \microtime(true); - $time = DateTime::now(); - - $enqueueDiff = $lastEnqueueUpdate === null ? 0 : $timerStart - $lastEnqueueUpdate; - $timeFrame = DateTime::addSeconds(new \DateTime(), self::FUNCTION_ENQUEUE_TIMER - $enqueueDiff); - - Console::log("Enqueue tick: started at: $time (with diff $enqueueDiff)"); - - $total = 0; - - $delayedExecutions = []; // Group executions with same delay to share one coroutine - - foreach ($schedules as $key => $schedule) { - $cron = new CronExpression($schedule['schedule']); - $nextDate = $cron->getNextRunDate(); - $next = DateTime::format($nextDate); - - $currentTick = $next < $timeFrame; - - if (!$currentTick) { - continue; - } - - $total++; - - $promiseStart = \time(); // in seconds - $executionStart = $nextDate->getTimestamp(); // in seconds - $delay = $executionStart - $promiseStart; // Time to wait from now until execution needs to be queued - - if (!isset($delayedExecutions[$delay])) { - $delayedExecutions[$delay] = []; - } - - $delayedExecutions[$delay][] = $key; - } - - foreach ($delayedExecutions as $delay => $scheduleKeys) { - \go(function () use ($delay, $schedules, $scheduleKeys, $pools) { - \sleep($delay); // in seconds - - $queue = $pools->get('queue')->pop(); - $connection = $queue->getResource(); - - foreach ($scheduleKeys as $scheduleKey) { - // Ensure schedule was not deleted - if (!isset($schedules[$scheduleKey])) { - return; - } - - $schedule = $schedules[$scheduleKey]; - - $functions = new Func($connection); - - $functions - ->setType('schedule') - ->setFunction($schedule['function']) - ->setMethod('POST') - ->setPath('/') - ->setProject($schedule['project']) - ->trigger(); - } - - $queue->reclaim(); - }); - } - - $timerEnd = \microtime(true); - $lastEnqueueUpdate = $timerStart; - Console::log("Enqueue tick: {$total} executions were enqueued in " . ($timerEnd - $timerStart) . " seconds"); - }; - - Timer::tick(self::FUNCTION_ENQUEUE_TIMER * 1000, fn() => $enqueueFunctions()); - $enqueueFunctions(); - } - ); - } -} diff --git a/src/Appwrite/Platform/Tasks/ScheduleBase.php b/src/Appwrite/Platform/Tasks/ScheduleBase.php new file mode 100644 index 0000000000..bb9a64ed62 --- /dev/null +++ b/src/Appwrite/Platform/Tasks/ScheduleBase.php @@ -0,0 +1,187 @@ +desc("Execute {$type}s scheduled in Appwrite") + ->inject('pools') + ->inject('dbForConsole') + ->inject('getProjectDB') + ->callback(fn(Group $pools, Database $dbForConsole, callable $getProjectDB) => $this->action($pools, $dbForConsole, $getProjectDB)); + } + + /** + * 1. Load all documents from 'schedules' collection to create local copy + * 2. Create timer that sync all changes from 'schedules' collection to local copy. Only reading changes thanks to 'resourceUpdatedAt' attribute + * 3. Create timer that prepares coroutines for soon-to-execute schedules. When it's ready, coroutine sleeps until exact time before sending request to worker. + */ + public function action(Group $pools, Database $dbForConsole, callable $getProjectDB): void + { + Console::title(\ucfirst(static::getSupportedResource()) . ' scheduler V1'); + Console::success(APP_NAME . ' ' . \ucfirst(static::getSupportedResource()) . ' scheduler v1 has started'); + + /** + * Extract only necessary attributes to lower memory used. + * + * @return array + * @throws Exception + * @var Document $schedule + */ + $getSchedule = function (Document $schedule) use ($dbForConsole, $getProjectDB): array { + $project = $dbForConsole->getDocument('projects', $schedule->getAttribute('projectId')); + + $resource = $getProjectDB($project)->getDocument( + $schedule->getAttribute('resourceCollection'), + $schedule->getAttribute('resourceId') + ); + + return [ + 'resourceId' => $schedule->getAttribute('resourceId'), + 'schedule' => $schedule->getAttribute('schedule'), + 'resourceUpdatedAt' => $schedule->getAttribute('resourceUpdatedAt'), + 'project' => $project, // TODO: @Meldiron Send only ID to worker to reduce memory usage here + 'resource' => $resource, // TODO: @Meldiron Send only ID to worker to reduce memory usage here + ]; + }; + + $lastSyncUpdate = DateTime::now(); + + $limit = 10_000; + $sum = $limit; + $total = 0; + $loadStart = \microtime(true); + $latestDocument = null; + + while ($sum === $limit) { + $paginationQueries = [Query::limit($limit)]; + + if ($latestDocument) { + $paginationQueries[] = Query::cursorAfter($latestDocument); + } + + $results = $dbForConsole->find('schedules', \array_merge($paginationQueries, [ + Query::equal('region', [App::getEnv('_APP_REGION', 'default')]), + Query::equal('resourceType', [static::getSupportedResource()]), + Query::equal('active', [true]), + ])); + + $sum = \count($results); + $total = $total + $sum; + + foreach ($results as $document) { + try { + $this->schedules[$document['resourceId']] = $getSchedule($document); + } catch (\Throwable $th) { + Console::error("Failed to load schedule for project {$document['projectId']} {$document['resourceCollection']} {$document['resourceId']}"); + Console::error($th->getMessage()); + } + } + + $latestDocument = \end($results); + } + + $pools->reclaim(); + + Console::success("{$total} resources were loaded in " . (\microtime(true) - $loadStart) . " seconds"); + + Console::success("Starting timers at " . DateTime::now()); + + run(function () use ($dbForConsole, &$lastSyncUpdate, $getSchedule, $pools) { + /** + * The timer synchronize $schedules copy with database collection. + */ + Timer::tick(static::UPDATE_TIMER * 1000, function () use ($dbForConsole, &$lastSyncUpdate, $getSchedule, $pools) { + $time = DateTime::now(); + $timerStart = \microtime(true); + + $limit = 1000; + $sum = $limit; + $total = 0; + $latestDocument = null; + + Console::log("Sync tick: Running at $time"); + + while ($sum === $limit) { + $paginationQueries = [Query::limit($limit)]; + + if ($latestDocument) { + $paginationQueries[] = Query::cursorAfter($latestDocument); + } + + $results = $dbForConsole->find('schedules', \array_merge($paginationQueries, [ + Query::equal('region', [App::getEnv('_APP_REGION', 'default')]), + Query::equal('resourceType', [static::getSupportedResource()]), + Query::greaterThanEqual('resourceUpdatedAt', $lastSyncUpdate), + ])); + + $sum = count($results); + $total = $total + $sum; + + foreach ($results as $document) { + $localDocument = $schedules[$document['resourceId']] ?? null; + + // Check if resource has been updated since last sync + $org = $localDocument !== null ? \strtotime($localDocument['resourceUpdatedAt']) : null; + $new = \strtotime($document['resourceUpdatedAt']); + + if (!$document['active']) { + Console::info("Removing: {$document['resourceId']}"); + unset($this->schedules[$document['resourceId']]); + } elseif ($new !== $org) { + Console::info("Updating: {$document['resourceId']}"); + $this->schedules[$document['resourceId']] = $getSchedule($document); + } + } + + $latestDocument = \end($results); + } + + $lastSyncUpdate = $time; + $timerEnd = \microtime(true); + + $pools->reclaim(); + + Console::log("Sync tick: {$total} schedules were updated in " . ($timerEnd - $timerStart) . " seconds"); + }); + + Timer::tick(static::ENQUEUE_TIMER * 1000, fn() => + $this->enqueueResources($pools, $dbForConsole)); + + $this->enqueueResources($pools, $dbForConsole); + }); + } +} diff --git a/src/Appwrite/Platform/Tasks/ScheduleFunctions.php b/src/Appwrite/Platform/Tasks/ScheduleFunctions.php new file mode 100644 index 0000000000..3128eb0538 --- /dev/null +++ b/src/Appwrite/Platform/Tasks/ScheduleFunctions.php @@ -0,0 +1,101 @@ +lastEnqueueUpdate === null ? 0 : $timerStart - $this->lastEnqueueUpdate; + $timeFrame = DateTime::addSeconds(new \DateTime(), static::ENQUEUE_TIMER - $enqueueDiff); + + Console::log("Enqueue tick: started at: $time (with diff $enqueueDiff)"); + + $total = 0; + + $delayedExecutions = []; // Group executions with same delay to share one coroutine + + foreach ($this->schedules as $key => $schedule) { + $cron = new CronExpression($schedule['schedule']); + $nextDate = $cron->getNextRunDate(); + $next = DateTime::format($nextDate); + + $currentTick = $next < $timeFrame; + + if (!$currentTick) { + continue; + } + + $total++; + + $promiseStart = \time(); // in seconds + $executionStart = $nextDate->getTimestamp(); // in seconds + $delay = $executionStart - $promiseStart; // Time to wait from now until execution needs to be queued + + if (!isset($delayedExecutions[$delay])) { + $delayedExecutions[$delay] = []; + } + + $delayedExecutions[$delay][] = $key; + } + + foreach ($delayedExecutions as $delay => $scheduleKeys) { + \go(function () use ($delay, $scheduleKeys, $pools) { + \sleep($delay); // in seconds + + $queue = $pools->get('queue')->pop(); + $connection = $queue->getResource(); + + foreach ($scheduleKeys as $scheduleKey) { + // Ensure schedule was not deleted + if (!isset($schedules[$scheduleKey])) { + return; + } + + $schedule = $schedules[$scheduleKey]; + + $queueForFunctions = new Func($connection); + + $queueForFunctions + ->setType('schedule') + ->setFunction($schedule['resource']) + ->setMethod('POST') + ->setPath('/') + ->setProject($schedule['project']) + ->trigger(); + } + + $queue->reclaim(); + }); + } + + $timerEnd = \microtime(true); + $this->lastEnqueueUpdate = $timerStart; + Console::log("Enqueue tick: {$total} executions were enqueued in " . ($timerEnd - $timerStart) . " seconds"); + } +} diff --git a/src/Appwrite/Platform/Tasks/ScheduleMessage.php b/src/Appwrite/Platform/Tasks/ScheduleMessage.php deleted file mode 100644 index 849df8587c..0000000000 --- a/src/Appwrite/Platform/Tasks/ScheduleMessage.php +++ /dev/null @@ -1,162 +0,0 @@ -desc('Execute functions scheduled in Appwrite') - ->inject('pools') - ->inject('dbForConsole') - ->inject('getProjectDB') - ->callback(fn (Group $pools, Database $dbForConsole, callable $getProjectDB) => $this->action($pools, $dbForConsole, $getProjectDB)); - } - - /** - * 1. Load all documents from 'schedules' collection to create local copy - * 2. Create timer that sync all changes from 'schedules' collection to local copy. Only reading changes thanks to 'resourceUpdatedAt' attribute - * 3. Create timer that prepares coroutines for soon-to-execute schedules. When it's ready, coroutime sleeps until exact time before sending request to worker. - */ - public function action(Group $pools, Database $dbForConsole, callable $getProjectDB): void - { - Console::title('Scheduler V1'); - Console::success(APP_NAME . ' Scheduler v1 has started'); - - $schedules = []; // Local copy of 'schedules' collection - $lastSyncUpdate = DateTime::now(); - - $limit = 10000; - $sum = $limit; - $total = 0; - $loadStart = \microtime(true); - $latestDocument = null; - - while ($sum === $limit) { - $paginationQueries = [Query::limit($limit)]; - if ($latestDocument !== null) { - $paginationQueries[] = Query::cursorAfter($latestDocument); - } - try { - $results = $dbForConsole->find('schedules', \array_merge($paginationQueries, [ - Query::lessThanEqual('schedule', DateTime::formatTz(DateTime::now())), - Query::equal('resourceType', ['message']), - Query::equal('active', [true]), - ])); - } catch (\Exception $e) { - var_dump($e->getTraceAsString()); - } - - $sum = count($results); - $total = $total + $sum; - foreach ($results as $schedule) { - $schedules[$schedule->getId()] = $schedule; - } - - $latestDocument = !empty(array_key_last($results)) ? $results[array_key_last($results)] : null; - } - - $pools->reclaim(); - - Console::success("{$total} message were loaded in " . (microtime(true) - $loadStart) . " seconds"); - - Console::success("Starting timers at " . DateTime::now()); - - run( - function () use ($dbForConsole, &$schedules, &$lastSyncUpdate, $pools) { - /** - * The timer synchronize $schedules copy with database collection. - */ - Timer::tick(self::MESSAGE_UPDATE_TIMER * 1000, function () use ($dbForConsole, &$schedules, &$lastSyncUpdate, $pools) { - $time = DateTime::now(); - $timerStart = \microtime(true); - - $limit = 1000; - $sum = $limit; - $total = 0; - $latestDocument = null; - - Console::log("Sync tick: Running at $time"); - - while ($sum === $limit) { - $paginationQueries = [Query::limit($limit)]; - if ($latestDocument !== null) { - $paginationQueries[] = Query::cursorAfter($latestDocument); - } - $results = $dbForConsole->find('schedules', \array_merge($paginationQueries, [ - Query::lessThanEqual('schedule', DateTime::formatTz(DateTime::now())), - Query::equal('resourceType', ['message']), - Query::equal('active', [true]), - ])); - $sum = \count($results); - $total = $total + $sum; - foreach ($results as $schedule) { - $schedules[$schedule->getId()] = $schedule; - } - - $latestDocument = !empty(array_key_last($results)) ? $results[array_key_last($results)] : null; - } - - $lastSyncUpdate = $time; - $timerEnd = \microtime(true); - - $pools->reclaim(); - - Console::log("Sync tick: {$total} schedules were updated in " . ($timerEnd - $timerStart) . " seconds"); - }); - - /** - * The timer to prepare soon-to-execute schedules. - */ - $enqueueMessages = function () use (&$schedules, $pools, $dbForConsole) { - foreach ($schedules as $scheduleId => $schedule) { - \go(function () use (&$schedules, $schedule, $pools, $dbForConsole) { - $queue = $pools->get('queue')->pop(); - $connection = $queue->getResource(); - $queueForMessaging = new Messaging($connection); - $queueForDeletes = new Delete($connection); - $project = $dbForConsole->getDocument('projects', $schedule->getAttribute('projectId')); - $queueForMessaging - ->setMessageId($schedule->getAttribute('resourceId')) - ->setProject($project) - ->trigger(); - $schedule->setAttribute('active', false); - $dbForConsole->updateDocument('schedules', $schedule->getId(), $schedule); - - $queueForDeletes - ->setType(DELETE_TYPE_SCHEDULES) - ->setDocument($schedule); - - $queue->reclaim(); - unset($schedules[$schedule->getId()]); - }); - } - }; - - Timer::tick(self::MESSAGE_ENQUEUE_TIMER * 1000, fn () => $enqueueMessages()); - $enqueueMessages(); - } - ); - } -} diff --git a/src/Appwrite/Platform/Tasks/ScheduleMessages.php b/src/Appwrite/Platform/Tasks/ScheduleMessages.php new file mode 100644 index 0000000000..6d938a8b4d --- /dev/null +++ b/src/Appwrite/Platform/Tasks/ScheduleMessages.php @@ -0,0 +1,57 @@ +schedules as $schedule) { + \go(function () use ($schedule, $pools, $dbForConsole) { + $queue = $pools->get('queue')->pop(); + $connection = $queue->getResource(); + $queueForMessaging = new Messaging($connection); + $queueForDeletes = new Delete($connection); + + $queueForMessaging + ->setMessageId($schedule['resourceId']) + ->setProject($schedule['project']) + ->trigger(); + + $queueForDeletes + ->setType(DELETE_TYPE_SCHEDULES) + ->setDocument($schedule) + ->trigger(); + + $queue->reclaim(); + + unset($this->schedules[$schedule->getId()]); + }); + } + } +} From 87a4d7db8823df7308520bc2e2bc7384805e854c Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 11 Jan 2024 16:08:40 +1300 Subject: [PATCH 08/30] Update service refs --- src/Appwrite/Platform/Services/Tasks.php | 56 +++++++++++----------- src/Appwrite/Platform/Services/Workers.php | 12 ++--- 2 files changed, 34 insertions(+), 34 deletions(-) diff --git a/src/Appwrite/Platform/Services/Tasks.php b/src/Appwrite/Platform/Services/Tasks.php index 51267e14fa..d1fb064999 100644 --- a/src/Appwrite/Platform/Services/Tasks.php +++ b/src/Appwrite/Platform/Services/Tasks.php @@ -2,27 +2,27 @@ namespace Appwrite\Platform\Services; -use Utopia\Platform\Service; +use Appwrite\Platform\Tasks\CalcTierStats; +use Appwrite\Platform\Tasks\DeleteOrphanedProjects; +use Appwrite\Platform\Tasks\DevGenerateTranslations; use Appwrite\Platform\Tasks\Doctor; +use Appwrite\Platform\Tasks\GetMigrationStats; +use Appwrite\Platform\Tasks\Hamster; use Appwrite\Platform\Tasks\Install; use Appwrite\Platform\Tasks\Maintenance; use Appwrite\Platform\Tasks\Migrate; -use Appwrite\Platform\Tasks\Schedule; +use Appwrite\Platform\Tasks\PatchRecreateRepositoriesDocuments; use Appwrite\Platform\Tasks\SDKs; -use Appwrite\Platform\Tasks\Specs; use Appwrite\Platform\Tasks\SSL; -use Appwrite\Platform\Tasks\Hamster; +use Appwrite\Platform\Tasks\ScheduleFunctions; +use Appwrite\Platform\Tasks\ScheduleMessages; +use Appwrite\Platform\Tasks\Specs; +use Appwrite\Platform\Tasks\Upgrade; use Appwrite\Platform\Tasks\Usage; use Appwrite\Platform\Tasks\Vars; use Appwrite\Platform\Tasks\Version; use Appwrite\Platform\Tasks\VolumeSync; -use Appwrite\Platform\Tasks\CalcTierStats; -use Appwrite\Platform\Tasks\Upgrade; -use Appwrite\Platform\Tasks\DeleteOrphanedProjects; -use Appwrite\Platform\Tasks\DevGenerateTranslations; -use Appwrite\Platform\Tasks\GetMigrationStats; -use Appwrite\Platform\Tasks\PatchRecreateRepositoriesDocuments; -use Appwrite\Platform\Tasks\ScheduleMessage; +use Utopia\Platform\Service; class Tasks extends Service { @@ -30,26 +30,26 @@ class Tasks extends Service { $this->type = self::TYPE_CLI; $this - ->addAction(Version::getName(), new Version()) - ->addAction(Usage::getName(), new Usage()) - ->addAction(Vars::getName(), new Vars()) - ->addAction(SSL::getName(), new SSL()) - ->addAction(Hamster::getName(), new Hamster()) - ->addAction(Doctor::getName(), new Doctor()) - ->addAction(Install::getName(), new Install()) - ->addAction(Upgrade::getName(), new Upgrade()) - ->addAction(Maintenance::getName(), new Maintenance()) - ->addAction(ScheduleMessage::getName(), new ScheduleMessage()) - ->addAction(Schedule::getName(), new Schedule()) - ->addAction(Migrate::getName(), new Migrate()) - ->addAction(SDKs::getName(), new SDKs()) - ->addAction(VolumeSync::getName(), new VolumeSync()) - ->addAction(Specs::getName(), new Specs()) ->addAction(CalcTierStats::getName(), new CalcTierStats()) ->addAction(DeleteOrphanedProjects::getName(), new DeleteOrphanedProjects()) - ->addAction(PatchRecreateRepositoriesDocuments::getName(), new PatchRecreateRepositoriesDocuments()) - ->addAction(GetMigrationStats::getName(), new GetMigrationStats()) ->addAction(DevGenerateTranslations::getName(), new DevGenerateTranslations()) + ->addAction(Doctor::getName(), new Doctor()) + ->addAction(GetMigrationStats::getName(), new GetMigrationStats()) + ->addAction(Hamster::getName(), new Hamster()) + ->addAction(Install::getName(), new Install()) + ->addAction(Maintenance::getName(), new Maintenance()) + ->addAction(Migrate::getName(), new Migrate()) + ->addAction(PatchRecreateRepositoriesDocuments::getName(), new PatchRecreateRepositoriesDocuments()) + ->addAction(SDKs::getName(), new SDKs()) + ->addAction(SSL::getName(), new SSL()) + ->addAction(ScheduleFunctions::getName(), new ScheduleFunctions()) + ->addAction(ScheduleMessages::getName(), new ScheduleMessages()) + ->addAction(Specs::getName(), new Specs()) + ->addAction(Upgrade::getName(), new Upgrade()) + ->addAction(Usage::getName(), new Usage()) + ->addAction(Vars::getName(), new Vars()) + ->addAction(Version::getName(), new Version()) + ->addAction(VolumeSync::getName(), new VolumeSync()) ; } diff --git a/src/Appwrite/Platform/Services/Workers.php b/src/Appwrite/Platform/Services/Workers.php index c5a0514760..ed6a5f44a0 100644 --- a/src/Appwrite/Platform/Services/Workers.php +++ b/src/Appwrite/Platform/Services/Workers.php @@ -22,16 +22,16 @@ class Workers extends Service $this->type = self::TYPE_WORKER; $this ->addAction(Audits::getName(), new Audits()) - ->addAction(Webhooks::getName(), new Webhooks()) - ->addAction(Mails::getName(), new Mails()) - ->addAction(Messaging::getName(), new Messaging()) + ->addAction(Builds::getName(), new Builds()) ->addAction(Certificates::getName(), new Certificates()) ->addAction(Databases::getName(), new Databases()) - ->addAction(Functions::getName(), new Functions()) - ->addAction(Builds::getName(), new Builds()) ->addAction(Deletes::getName(), new Deletes()) - ->addAction(Migrations::getName(), new Migrations()) + ->addAction(Functions::getName(), new Functions()) ->addAction(Hamster::getName(), new Hamster()) + ->addAction(Mails::getName(), new Mails()) + ->addAction(Messaging::getName(), new Messaging()) + ->addAction(Migrations::getName(), new Migrations()) + ->addAction(Webhooks::getName(), new Webhooks()) ; } From 1301031b48155237a8e4f13fe4fd3e96ab3954d1 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 11 Jan 2024 16:10:20 +1300 Subject: [PATCH 09/30] Update bin scripts --- Dockerfile | 32 ++++++++++++++++---------------- bin/schedule | 3 --- bin/schedule-functions | 3 +++ bin/schedule-message | 3 --- bin/schedule-messages | 3 +++ 5 files changed, 22 insertions(+), 22 deletions(-) delete mode 100644 bin/schedule create mode 100644 bin/schedule-functions delete mode 100644 bin/schedule-message create mode 100644 bin/schedule-messages diff --git a/Dockerfile b/Dockerfile index 190df719ba..19013ab0a0 100755 --- a/Dockerfile +++ b/Dockerfile @@ -76,42 +76,42 @@ RUN chmod +x /usr/local/bin/dev-generate-translations # Executables RUN chmod +x /usr/local/bin/doctor && \ - chmod +x /usr/local/bin/maintenance && \ - chmod +x /usr/local/bin/usage && \ chmod +x /usr/local/bin/install && \ - chmod +x /usr/local/bin/upgrade && \ + chmod +x /usr/local/bin/maintenance && \ chmod +x /usr/local/bin/migrate && \ chmod +x /usr/local/bin/realtime && \ - chmod +x /usr/local/bin/schedule && \ - chmod +x /usr/local/bin/schedule-message && \ + chmod +x /usr/local/bin/schedule-functions && \ + chmod +x /usr/local/bin/schedule-messages && \ chmod +x /usr/local/bin/sdks && \ chmod +x /usr/local/bin/specs && \ chmod +x /usr/local/bin/ssl && \ chmod +x /usr/local/bin/test && \ + chmod +x /usr/local/bin/upgrade && \ + chmod +x /usr/local/bin/usage && \ chmod +x /usr/local/bin/vars && \ chmod +x /usr/local/bin/worker-audits && \ + chmod +x /usr/local/bin/worker-builds && \ chmod +x /usr/local/bin/worker-certificates && \ chmod +x /usr/local/bin/worker-databases && \ chmod +x /usr/local/bin/worker-deletes && \ chmod +x /usr/local/bin/worker-functions && \ - chmod +x /usr/local/bin/worker-builds && \ + chmod +x /usr/local/bin/worker-hamster && \ chmod +x /usr/local/bin/worker-mails && \ chmod +x /usr/local/bin/worker-messaging && \ - chmod +x /usr/local/bin/worker-webhooks && \ chmod +x /usr/local/bin/worker-migrations && \ - chmod +x /usr/local/bin/worker-hamster + chmod +x /usr/local/bin/worker-webhooks # Cloud Executabless -RUN chmod +x /usr/local/bin/hamster && \ - chmod +x /usr/local/bin/volume-sync && \ +RUN chmod +x /usr/local/bin/calc-tier-stats && \ + chmod +x /usr/local/bin/calc-users-stats && \ + chmod +x /usr/local/bin/clear-card-cache && \ + chmod +x /usr/local/bin/delete-orphaned-projects && \ + chmod +x /usr/local/bin/get-migration-stats && \ + chmod +x /usr/local/bin/hamster && \ + chmod +x /usr/local/bin/patch-delete-project-collections && \ chmod +x /usr/local/bin/patch-delete-schedule-updated-at-attribute && \ chmod +x /usr/local/bin/patch-recreate-repositories-documents && \ - chmod +x /usr/local/bin/patch-delete-project-collections && \ - chmod +x /usr/local/bin/delete-orphaned-projects && \ - chmod +x /usr/local/bin/clear-card-cache && \ - chmod +x /usr/local/bin/calc-users-stats && \ - chmod +x /usr/local/bin/calc-tier-stats && \ - chmod +x /usr/local/bin/get-migration-stats + chmod +x /usr/local/bin/volume-sync # Letsencrypt Permissions RUN mkdir -p /etc/letsencrypt/live/ && chmod -Rf 755 /etc/letsencrypt/live/ diff --git a/bin/schedule b/bin/schedule deleted file mode 100644 index ddd1ea7f35..0000000000 --- a/bin/schedule +++ /dev/null @@ -1,3 +0,0 @@ -#!/bin/sh - -php /usr/src/code/app/cli.php schedule $@ \ No newline at end of file diff --git a/bin/schedule-functions b/bin/schedule-functions new file mode 100644 index 0000000000..10edbe8226 --- /dev/null +++ b/bin/schedule-functions @@ -0,0 +1,3 @@ +#!/bin/sh + +php /usr/src/code/app/cli.php schedule-functions $@ \ No newline at end of file diff --git a/bin/schedule-message b/bin/schedule-message deleted file mode 100644 index 62e0fdbe6e..0000000000 --- a/bin/schedule-message +++ /dev/null @@ -1,3 +0,0 @@ -#!/bin/sh - -php /usr/src/code/app/cli.php schedule-message $@ \ No newline at end of file diff --git a/bin/schedule-messages b/bin/schedule-messages new file mode 100644 index 0000000000..fa7219f6ea --- /dev/null +++ b/bin/schedule-messages @@ -0,0 +1,3 @@ +#!/bin/sh + +php /usr/src/code/app/cli.php schedule-messages $@ \ No newline at end of file From 38e9aefa2fd18b33ecd99fcd11be1a9bce824833 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 11 Jan 2024 16:14:38 +1300 Subject: [PATCH 10/30] Update compose --- app/views/install/compose.phtml | 31 ++++++++++++++++++++++++++++--- docker-compose.yml | 12 ++++++------ 2 files changed, 34 insertions(+), 9 deletions(-) diff --git a/app/views/install/compose.phtml b/app/views/install/compose.phtml index 252b5b6bd7..c692ac22fb 100644 --- a/app/views/install/compose.phtml +++ b/app/views/install/compose.phtml @@ -632,10 +632,35 @@ services: - _APP_LOGGING_PROVIDER - _APP_LOGGING_CONFIG - appwrite-schedule: + appwrite-scheduler-functions: image: /: - entrypoint: schedule - container_name: appwrite-schedule + entrypoint: schedule-functions + container_name: appwrite-scheduler-functions + <<: *x-logging + restart: unless-stopped + networks: + - appwrite + depends_on: + - mariadb + - redis + environment: + - _APP_ENV + - _APP_WORKER_PER_CORE + - _APP_OPENSSL_KEY_V1 + - _APP_REDIS_HOST + - _APP_REDIS_PORT + - _APP_REDIS_USER + - _APP_REDIS_PASS + - _APP_DB_HOST + - _APP_DB_PORT + - _APP_DB_SCHEMA + - _APP_DB_USER + - _APP_DB_PASS + + appwrite-scheduler-messages: + image: /: + entrypoint: schedule-messages + container_name: appwrite-scheduler-messages <<: *x-logging restart: unless-stopped networks: diff --git a/docker-compose.yml b/docker-compose.yml index 892b280798..66d9091157 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -690,10 +690,10 @@ services: - _APP_LOGGING_PROVIDER - _APP_LOGGING_CONFIG - appwrite-schedule: - entrypoint: schedule + appwrite-scheduler-functions: + entrypoint: schedule-functions <<: *x-logging - container_name: appwrite-schedule + container_name: appwrite-scheduler-functions image: appwrite-dev networks: - appwrite @@ -717,10 +717,10 @@ services: - _APP_DB_USER - _APP_DB_PASS - appwrite-schedule-message: - entrypoint: schedule-message + appwrite-scheduler-messages: + entrypoint: schedule-messages <<: *x-logging - container_name: appwrite-schedule-message + container_name: appwrite-scheduler-messages image: appwrite-dev networks: - appwrite From 62304f817d15c6b3dbfea2ec95dda0d2adebe21c Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 11 Jan 2024 16:15:11 +1300 Subject: [PATCH 11/30] Fix maintenance worker --- src/Appwrite/Platform/Tasks/Maintenance.php | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/Appwrite/Platform/Tasks/Maintenance.php b/src/Appwrite/Platform/Tasks/Maintenance.php index c9928dd163..58e3228d27 100644 --- a/src/Appwrite/Platform/Tasks/Maintenance.php +++ b/src/Appwrite/Platform/Tasks/Maintenance.php @@ -61,7 +61,7 @@ class Maintenance extends Action private function notifyDeleteExecutionLogs(int $interval, Delete $queueForDeletes): void { - ($queueForDeletes) + $queueForDeletes ->setType(DELETE_TYPE_EXECUTIONS) ->setDatetime(DateTime::addSeconds(new \DateTime(), -1 * $interval)) ->trigger(); @@ -69,7 +69,7 @@ class Maintenance extends Action private function notifyDeleteAbuseLogs(int $interval, Delete $queueForDeletes): void { - ($queueForDeletes) + $queueForDeletes ->setType(DELETE_TYPE_ABUSE) ->setDatetime(DateTime::addSeconds(new \DateTime(), -1 * $interval)) ->trigger(); @@ -77,7 +77,7 @@ class Maintenance extends Action private function notifyDeleteAuditLogs(int $interval, Delete $queueForDeletes): void { - ($queueForDeletes) + $queueForDeletes ->setType(DELETE_TYPE_AUDIT) ->setDatetime(DateTime::addSeconds(new \DateTime(), -1 * $interval)) ->trigger(); @@ -85,7 +85,7 @@ class Maintenance extends Action private function notifyDeleteUsageStats(int $usageStatsRetentionHourly, Delete $queueForDeletes): void { - ($queueFor) + $queueForDeletes ->setType(DELETE_TYPE_USAGE) ->setUsageRetentionHourlyDateTime(DateTime::addSeconds(new \DateTime(), -1 * $usageStatsRetentionHourly)) ->trigger(); @@ -93,7 +93,7 @@ class Maintenance extends Action private function notifyDeleteConnections(Delete $queueForDeletes): void { - ($queueForDeletes) + $queueForDeletes ->setType(DELETE_TYPE_REALTIME) ->setDatetime(DateTime::addSeconds(new \DateTime(), -60)) ->trigger(); @@ -101,7 +101,7 @@ class Maintenance extends Action private function notifyDeleteExpiredSessions(Delete $queueForDeletes): void { - ($queueForDeletes) + $queueForDeletes ->setType(DELETE_TYPE_SESSIONS) ->trigger(); } From 449e8cc06c4037f38ab1561703a558934d81615c Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 11 Jan 2024 19:32:24 +1300 Subject: [PATCH 12/30] Fix function schedules --- src/Appwrite/Platform/Tasks/ScheduleFunctions.php | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/Appwrite/Platform/Tasks/ScheduleFunctions.php b/src/Appwrite/Platform/Tasks/ScheduleFunctions.php index 3128eb0538..e2c278714f 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleFunctions.php +++ b/src/Appwrite/Platform/Tasks/ScheduleFunctions.php @@ -23,7 +23,7 @@ class ScheduleFunctions extends ScheduleBase public static function getSupportedResource(): string { - return 'message'; + return 'function'; } protected function enqueueResources(Group $pools, Database $dbForConsole): void @@ -73,11 +73,11 @@ class ScheduleFunctions extends ScheduleBase foreach ($scheduleKeys as $scheduleKey) { // Ensure schedule was not deleted - if (!isset($schedules[$scheduleKey])) { + if (!\array_key_exists($scheduleKey, $this->schedules)) { return; } - $schedule = $schedules[$scheduleKey]; + $schedule = $this->schedules[$scheduleKey]; $queueForFunctions = new Func($connection); @@ -95,7 +95,10 @@ class ScheduleFunctions extends ScheduleBase } $timerEnd = \microtime(true); - $this->lastEnqueueUpdate = $timerStart; + + // TODO: This was a bug before because it wasn't passed by reference, enabling it breaks scheduling + //$this->lastEnqueueUpdate = $timerStart; + Console::log("Enqueue tick: {$total} executions were enqueued in " . ($timerEnd - $timerStart) . " seconds"); } } From cb3e354cf150f2ac527c1ca3e0f615aedc7b5d67 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 11 Jan 2024 19:47:55 +1300 Subject: [PATCH 13/30] Fix tests --- app/controllers/api/projects.php | 48 ------------------- .../e2e/Services/Messaging/MessagingBase.php | 6 ++- 2 files changed, 4 insertions(+), 50 deletions(-) diff --git a/app/controllers/api/projects.php b/app/controllers/api/projects.php index 38d23b2d2d..fe441e0e8c 100644 --- a/app/controllers/api/projects.php +++ b/app/controllers/api/projects.php @@ -216,54 +216,6 @@ App::post('/v1/projects') } $dbForProject->createCollection($key, $attributes, $indexes); } - $emailProvider = new Document([ - '$id' => ID::custom('mock-email-provider'), - 'name' => 'mock', - 'provider' => 'mock', - 'type' => MESSAGE_TYPE_EMAIL, - 'enabled' => true, - 'credentials' => [ - 'username' => 'username', - 'password' => 'password' - ], - 'options' => [ - 'from' => 'sender-email' - ], - ]); - $smsProvider = new Document([ - '$id' => ID::custom('mock-sms-provider'), - 'name' => 'mock', - 'provider' => 'mock', - 'type' => MESSAGE_TYPE_SMS, - 'enabled' => true, - 'credentials' => [ - 'username' => 'username', - 'password' => 'password' - ], - 'options' => [ - 'from' => 'sender-email' - ], - ]); - $pushProvider = new Document([ - '$id' => ID::custom('mock-push-provider'), - 'name' => 'mock', - 'provider' => 'mock', - 'type' => MESSAGE_TYPE_PUSH, - 'enabled' => true, - 'credentials' => [ - 'username' => 'username', - 'password' => 'password' - ], - 'options' => [ - 'from' => 'sender-email' - ], - ]); - $dbForProject->createDocument('providers', $emailProvider); - - - $dbForProject->createDocument('providers', $smsProvider); - - $dbForProject->createDocument('providers', $pushProvider); $response ->setStatusCode(Response::STATUS_CODE_CREATED) diff --git a/tests/e2e/Services/Messaging/MessagingBase.php b/tests/e2e/Services/Messaging/MessagingBase.php index accf49b77f..a93faacd12 100644 --- a/tests/e2e/Services/Messaging/MessagingBase.php +++ b/tests/e2e/Services/Messaging/MessagingBase.php @@ -608,7 +608,7 @@ trait MessagingBase 'name' => 'Sendgrid-provider', 'apiKey' => $apiKey, 'fromName' => $fromName, - 'fromEmail' => $fromEmail + 'fromEmail' => $fromEmail, 'enabled' => true, ]); @@ -765,7 +765,8 @@ trait MessagingBase 'name' => 'Msg91Sender', 'senderId' => $senderId, 'authKey' => $authKey, - 'from' => $from + 'from' => $from, + 'enabled' => true, ]); $this->assertEquals(201, $provider['headers']['status-code']); @@ -926,6 +927,7 @@ trait MessagingBase 'providerId' => ID::unique(), 'name' => 'FCM-1', 'serviceAccountJSON' => $serviceAccountJSON, + 'enabled' => true, ]); $this->assertEquals(201, $provider['headers']['status-code']); From e21e2c70eaeb0b321a5bb2c6c4d03de99c5e8fcf Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Mon, 15 Jan 2024 18:18:09 +1300 Subject: [PATCH 14/30] Remove redundant index --- app/config/collections.php | 7 ------- 1 file changed, 7 deletions(-) diff --git a/app/config/collections.php b/app/config/collections.php index c82b6b671e..979a267fa4 100644 --- a/app/config/collections.php +++ b/app/config/collections.php @@ -4246,13 +4246,6 @@ $consoleCollections = array_merge([ 'lengths' => [], 'orders' => [], ], - [ - '$id' => ID::custom('_key_schedule_resourceType_active_resourceUpdatedAt'), - 'type' => Database::INDEX_KEY, - 'attributes' => ['schedule', 'resourceType', 'active', 'resourceUpdatedAt'], - 'lengths' => [], - 'orders' => [], - ] ], ], From 189406635ac728ca15aaf6a39921df1fdb8d7962 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Mon, 15 Jan 2024 18:25:34 +1300 Subject: [PATCH 15/30] Use internal ID for subquery --- app/init.php | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/app/init.php b/app/init.php index 3777028015..e08cca1954 100644 --- a/app/init.php +++ b/app/init.php @@ -545,15 +545,16 @@ Database::addFilter( }, function (mixed $value, Document $document, Database $database) { $targetIds = Authorization::skip(fn () => \array_map( - fn ($document) => $document->getAttribute('targetId'), - $database - ->find('subscribers', [ + fn ($document) => $document->getAttribute('targetInternalId'), + $database->find('subscribers', [ Query::equal('topicInternalId', [$document->getInternalId()]), Query::limit(APP_LIMIT_SUBSCRIBERS_SUBQUERY) ]) )); if (\count($targetIds) > 0) { - return $database->find('targets', [Query::equal('$id', $targetIds)]); + return $database->find('targets', [ + Query::equal('$internalId', $targetIds) + ]); } return []; } From 78fe9ebb374f174155607eebf79e3c3d1b16a98c Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Mon, 15 Jan 2024 18:26:30 +1300 Subject: [PATCH 16/30] Fix invalid coalesce --- src/Appwrite/Platform/Workers/Deletes.php | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/src/Appwrite/Platform/Workers/Deletes.php b/src/Appwrite/Platform/Workers/Deletes.php index 14703911ac..2283ee873c 100644 --- a/src/Appwrite/Platform/Workers/Deletes.php +++ b/src/Appwrite/Platform/Workers/Deletes.php @@ -20,6 +20,7 @@ use Utopia\Database\Exception\Authorization; use Utopia\Database\Exception\Conflict; use Utopia\Database\Exception\Restricted; use Utopia\Database\Exception\Structure; +use Utopia\Database\Exception as DatabaseException; use Utopia\Database\Query; use Utopia\Platform\Action; use Utopia\Queue\Message; @@ -163,9 +164,13 @@ class Deletes extends Action * @param Database $dbForConsole * @param callable $getProjectDB * @param string $datetime + * @param Document|null $document * @return void * @throws Authorization - * @throws Throwable + * @throws Conflict + * @throws Restricted + * @throws Structure + * @throws DatabaseException */ private function deleteSchedules(Database $dbForConsole, callable $getProjectDB, string $datetime, ?Document $document = null): void { @@ -173,7 +178,7 @@ class Deletes extends Action 'schedules', [ Query::equal('region', [App::getEnv('_APP_REGION', 'default')]), - Query::equal('resourceType', [$document ?? $document->getAttribute('resourceType')]), + Query::equal('resourceType', [$document->getAttribute('resourceType')]), Query::lessThanEqual('resourceUpdatedAt', $datetime), Query::equal('active', [false]), ], From 099094f719e0308d1447d3fa284f4bb48809f432 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Mon, 15 Jan 2024 18:27:48 +1300 Subject: [PATCH 17/30] Fix target fetch limits in worker --- src/Appwrite/Platform/Workers/Messaging.php | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/Appwrite/Platform/Workers/Messaging.php b/src/Appwrite/Platform/Workers/Messaging.php index 2d8770e29f..b5ba7142b1 100644 --- a/src/Appwrite/Platform/Workers/Messaging.php +++ b/src/Appwrite/Platform/Workers/Messaging.php @@ -95,7 +95,7 @@ class Messaging extends Action if (\count($topicIds) > 0) { $topics = $dbForProject->find('topics', [ Query::equal('$id', $topicIds), - Query::limit(APP_LIMIT_SUBSCRIBERS_SUBQUERY) + Query::limit($topicIds) ]); foreach ($topics as $topic) { $targets = \array_filter($topic->getAttribute('targets'), fn(Document $target) => @@ -107,7 +107,7 @@ class Messaging extends Action if (\count($userIds) > 0) { $users = $dbForProject->find('users', [ Query::equal('$id', $userIds), - Query::limit(APP_LIMIT_SUBSCRIBERS_SUBQUERY) + Query::limit($userIds) ]); foreach ($users as $user) { $targets = \array_filter($user->getAttribute('targets'), fn(Document $target) => @@ -119,7 +119,7 @@ class Messaging extends Action if (\count($targetIds) > 0) { $targets = $dbForProject->find('targets', [ Query::equal('$id', $targetIds), - Query::limit(APP_LIMIT_SUBSCRIBERS_SUBQUERY) + Query::limit($targetIds) ]); $recipients = \array_merge($recipients, $targets); } From a2d0385ebab397f477dfa6218bbd4376d34498c7 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Mon, 15 Jan 2024 18:28:40 +1300 Subject: [PATCH 18/30] Check more failure cases --- src/Appwrite/Platform/Workers/Messaging.php | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/Appwrite/Platform/Workers/Messaging.php b/src/Appwrite/Platform/Workers/Messaging.php index b5ba7142b1..865dd71c24 100644 --- a/src/Appwrite/Platform/Workers/Messaging.php +++ b/src/Appwrite/Platform/Workers/Messaging.php @@ -124,11 +124,21 @@ class Messaging extends Action $recipients = \array_merge($recipients, $targets); } + if (empty($recipients)) { + Console::error('No valid recipients found.'); + return; + } + $fallback = $dbForProject->findOne('providers', [ Query::equal('enabled', [true]), Query::equal('type', [$recipients[0]->getAttribute('providerType')]), ]); + if ($fallback === false) { + Console::error('No fallback provider found.'); + return; + } + /** * @var array> $identifiers */ From 1666ba1645076bc7e1bfbc61c83cc16d1e6dfdcb Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Mon, 15 Jan 2024 18:43:17 +1300 Subject: [PATCH 19/30] Fix missing time check --- src/Appwrite/Platform/Tasks/ScheduleBase.php | 6 ++++-- src/Appwrite/Platform/Tasks/ScheduleMessages.php | 11 +++++++++-- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/src/Appwrite/Platform/Tasks/ScheduleBase.php b/src/Appwrite/Platform/Tasks/ScheduleBase.php index bb9a64ed62..ed42a45e4e 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleBase.php +++ b/src/Appwrite/Platform/Tasks/ScheduleBase.php @@ -178,8 +178,10 @@ abstract class ScheduleBase extends Action Console::log("Sync tick: {$total} schedules were updated in " . ($timerEnd - $timerStart) . " seconds"); }); - Timer::tick(static::ENQUEUE_TIMER * 1000, fn() => - $this->enqueueResources($pools, $dbForConsole)); + Timer::tick( + static::ENQUEUE_TIMER * 1000, + fn() => $this->enqueueResources($pools, $dbForConsole) + ); $this->enqueueResources($pools, $dbForConsole); }); diff --git a/src/Appwrite/Platform/Tasks/ScheduleMessages.php b/src/Appwrite/Platform/Tasks/ScheduleMessages.php index 6d938a8b4d..9ede91279d 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleMessages.php +++ b/src/Appwrite/Platform/Tasks/ScheduleMessages.php @@ -16,7 +16,7 @@ use function Swoole\Coroutine\run; class ScheduleMessages extends ScheduleBase { - public const UPDATE_TIMER = 3; // seconds + public const UPDATE_TIMER = 10; // seconds public const ENQUEUE_TIMER = 60; // seconds public static function getName(): string @@ -32,6 +32,13 @@ class ScheduleMessages extends ScheduleBase protected function enqueueResources(Group $pools, Database $dbForConsole): void { foreach ($this->schedules as $schedule) { + $now = DateTime::now(); + $scheduledAt = DateTime::formatTz($schedule['scheduledAt']); + + if ($scheduledAt > $now) { + continue; + } + \go(function () use ($schedule, $pools, $dbForConsole) { $queue = $pools->get('queue')->pop(); $connection = $queue->getResource(); @@ -50,7 +57,7 @@ class ScheduleMessages extends ScheduleBase $queue->reclaim(); - unset($this->schedules[$schedule->getId()]); + unset($this->schedules[$schedule['resourceId']]); }); } } From f25dd3276762dc349a912bfef9099ef90cb36493 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Mon, 15 Jan 2024 19:00:41 +1300 Subject: [PATCH 20/30] Fix delete callback --- src/Appwrite/Platform/Workers/Deletes.php | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/src/Appwrite/Platform/Workers/Deletes.php b/src/Appwrite/Platform/Workers/Deletes.php index 2283ee873c..6ed0c5d496 100644 --- a/src/Appwrite/Platform/Workers/Deletes.php +++ b/src/Appwrite/Platform/Workers/Deletes.php @@ -192,11 +192,22 @@ class Deletes extends Action return; } - $function = $getProjectDB($project)->getDocument('functions', $document->getAttribute('resourceId')); + $resource = $getProjectDB($project)->getDocument( + $document->getAttribute('resourceCollection'), + $document->getAttribute('resourceId') + ); - if ($function->isEmpty()) { + $delete = true; + + switch ($document->getAttribute('resourceType')) { + case 'function': + $delete = $resource->isEmpty(); + break; + } + + if ($delete) { $dbForConsole->deleteDocument('schedules', $document->getId()); - Console::success('Deleting schedule for function ' . $document->getAttribute('resourceId')); + Console::success('Deleting schedule for ' . $document->getAttribute('resourceType') . ' ' . $document->getAttribute('resourceId')); } } ); From a05a5da9bb9dea010658be158c357fd390bb27d3 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Mon, 15 Jan 2024 19:02:54 +1300 Subject: [PATCH 21/30] Remove todo --- src/Appwrite/Platform/Workers/Messaging.php | 1 - 1 file changed, 1 deletion(-) diff --git a/src/Appwrite/Platform/Workers/Messaging.php b/src/Appwrite/Platform/Workers/Messaging.php index 865dd71c24..d57e1837d7 100644 --- a/src/Appwrite/Platform/Workers/Messaging.php +++ b/src/Appwrite/Platform/Workers/Messaging.php @@ -265,7 +265,6 @@ class Messaging extends Action $message->setAttribute('deliveryErrors', $deliveryErrors); if (\count($message->getAttribute('deliveryErrors')) > 0) { - // TODO: Does this make sense? Only some of the messages might have failed, but we mark the whole message as failed. $message->setAttribute('status', 'failed'); } else { $message->setAttribute('status', 'sent'); From ce78d8c473f0116f0dbe3e593641ff1394df048f Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Mon, 15 Jan 2024 19:29:19 +1300 Subject: [PATCH 22/30] Set schedule inactive after message send so deletes worker picks it up --- src/Appwrite/Platform/Tasks/ScheduleBase.php | 1 + src/Appwrite/Platform/Tasks/ScheduleMessages.php | 8 ++++++++ src/Appwrite/Platform/Workers/Messaging.php | 2 +- 3 files changed, 10 insertions(+), 1 deletion(-) diff --git a/src/Appwrite/Platform/Tasks/ScheduleBase.php b/src/Appwrite/Platform/Tasks/ScheduleBase.php index ed42a45e4e..1ec8e471bb 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleBase.php +++ b/src/Appwrite/Platform/Tasks/ScheduleBase.php @@ -70,6 +70,7 @@ abstract class ScheduleBase extends Action ); return [ + '$id' => $schedule->getId(), 'resourceId' => $schedule->getAttribute('resourceId'), 'schedule' => $schedule->getAttribute('schedule'), 'resourceUpdatedAt' => $schedule->getAttribute('resourceUpdatedAt'), diff --git a/src/Appwrite/Platform/Tasks/ScheduleMessages.php b/src/Appwrite/Platform/Tasks/ScheduleMessages.php index 9ede91279d..ee11578fea 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleMessages.php +++ b/src/Appwrite/Platform/Tasks/ScheduleMessages.php @@ -50,6 +50,14 @@ class ScheduleMessages extends ScheduleBase ->setProject($schedule['project']) ->trigger(); + $dbForConsole->updateDocument( + 'schedules', + $schedule['$id'], + $dbForConsole + ->getDocument('schedules', $schedule['$id']) + ->setAttribute('active', false) + ); + $queueForDeletes ->setType(DELETE_TYPE_SCHEDULES) ->setDocument($schedule) diff --git a/src/Appwrite/Platform/Workers/Messaging.php b/src/Appwrite/Platform/Workers/Messaging.php index d57e1837d7..9b7bda57bb 100644 --- a/src/Appwrite/Platform/Workers/Messaging.php +++ b/src/Appwrite/Platform/Workers/Messaging.php @@ -134,7 +134,7 @@ class Messaging extends Action Query::equal('type', [$recipients[0]->getAttribute('providerType')]), ]); - if ($fallback === false) { + if ($fallback === false || $fallback->isEmpty()) { Console::error('No fallback provider found.'); return; } From ae60089413c608e155d6f9a2dc2dee495eb5bfe0 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Mon, 15 Jan 2024 19:31:42 +1300 Subject: [PATCH 23/30] Inline update --- src/Appwrite/Platform/Tasks/ScheduleMessages.php | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/Appwrite/Platform/Tasks/ScheduleMessages.php b/src/Appwrite/Platform/Tasks/ScheduleMessages.php index ee11578fea..471aeb8f2a 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleMessages.php +++ b/src/Appwrite/Platform/Tasks/ScheduleMessages.php @@ -4,6 +4,7 @@ namespace Appwrite\Platform\Tasks; use Appwrite\Event\Delete; use Swoole\Timer; +use Utopia\Database\Document; use Utopia\Platform\Action; use Utopia\CLI\Console; use Utopia\Database\DateTime; @@ -53,9 +54,7 @@ class ScheduleMessages extends ScheduleBase $dbForConsole->updateDocument( 'schedules', $schedule['$id'], - $dbForConsole - ->getDocument('schedules', $schedule['$id']) - ->setAttribute('active', false) + new Document(['active' => false]) ); $queueForDeletes From be9728937a5eb6bab618fb4dd5b026fabce92fc5 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Wed, 17 Jan 2024 14:35:12 +1300 Subject: [PATCH 24/30] Update src/Appwrite/Platform/Tasks/ScheduleMessages.php Co-authored-by: Steven Nguyen <1477010+stnguyen90@users.noreply.github.com> --- src/Appwrite/Platform/Tasks/ScheduleMessages.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Appwrite/Platform/Tasks/ScheduleMessages.php b/src/Appwrite/Platform/Tasks/ScheduleMessages.php index 471aeb8f2a..19fc426637 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleMessages.php +++ b/src/Appwrite/Platform/Tasks/ScheduleMessages.php @@ -34,7 +34,7 @@ class ScheduleMessages extends ScheduleBase { foreach ($this->schedules as $schedule) { $now = DateTime::now(); - $scheduledAt = DateTime::formatTz($schedule['scheduledAt']); + $scheduledAt = DateTime::formatTz($schedule['schedule']); if ($scheduledAt > $now) { continue; From 0ffe1d5346887e546bf3655fee8f454a1f439b93 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Wed, 17 Jan 2024 14:54:25 +1300 Subject: [PATCH 25/30] Fix limits --- src/Appwrite/Platform/Workers/Messaging.php | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/Appwrite/Platform/Workers/Messaging.php b/src/Appwrite/Platform/Workers/Messaging.php index d0e47d1133..3267a9034e 100644 --- a/src/Appwrite/Platform/Workers/Messaging.php +++ b/src/Appwrite/Platform/Workers/Messaging.php @@ -97,7 +97,7 @@ class Messaging extends Action if (\count($topicIds) > 0) { $topics = $dbForProject->find('topics', [ Query::equal('$id', $topicIds), - Query::limit($topicIds) + Query::limit(\count($topicIds)), ]); foreach ($topics as $topic) { $targets = \array_filter($topic->getAttribute('targets'), fn(Document $target) => @@ -109,7 +109,7 @@ class Messaging extends Action if (\count($userIds) > 0) { $users = $dbForProject->find('users', [ Query::equal('$id', $userIds), - Query::limit($userIds) + Query::limit(\count($userIds)), ]); foreach ($users as $user) { $targets = \array_filter($user->getAttribute('targets'), fn(Document $target) => @@ -121,7 +121,7 @@ class Messaging extends Action if (\count($targetIds) > 0) { $targets = $dbForProject->find('targets', [ Query::equal('$id', $targetIds), - Query::limit($targetIds) + Query::limit(\count($targetIds)), ]); $recipients = \array_merge($recipients, $targets); } From aee1d4406289c92651e3af3396ba79b5d0c97373 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Wed, 17 Jan 2024 14:57:29 +1300 Subject: [PATCH 26/30] Set failed and delivery errors for failure cases --- src/Appwrite/Platform/Workers/Messaging.php | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/src/Appwrite/Platform/Workers/Messaging.php b/src/Appwrite/Platform/Workers/Messaging.php index 3267a9034e..cf05b42d43 100644 --- a/src/Appwrite/Platform/Workers/Messaging.php +++ b/src/Appwrite/Platform/Workers/Messaging.php @@ -127,8 +127,12 @@ class Messaging extends Action } if (empty($recipients)) { - Console::error('No valid recipients found.'); - return; + $dbForProject->updateDocument('messages', $message->getId(), $message->setAttributes([ + 'status' => 'failed', + 'deliveryErrors' => ['No valid recipients found.'] + ])); + + throw new \Exception('No valid recipients found.'); } $fallback = $dbForProject->findOne('providers', [ @@ -137,8 +141,12 @@ class Messaging extends Action ]); if ($fallback === false || $fallback->isEmpty()) { - Console::error('No fallback provider found.'); - return; + $dbForProject->updateDocument('messages', $message->getId(), $message->setAttributes([ + 'status' => 'failed', + 'deliveryErrors' => ['No fallback provider found.'] + ])); + + throw new \Exception('No fallback provider found.'); } /** From fe0af8e2cd33859ba636b943471fb7e45a06dd06 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Wed, 17 Jan 2024 15:06:10 +1300 Subject: [PATCH 27/30] Add missing provider filter for targets --- src/Appwrite/Platform/Workers/Messaging.php | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/Appwrite/Platform/Workers/Messaging.php b/src/Appwrite/Platform/Workers/Messaging.php index cf05b42d43..ca33b26d69 100644 --- a/src/Appwrite/Platform/Workers/Messaging.php +++ b/src/Appwrite/Platform/Workers/Messaging.php @@ -123,6 +123,8 @@ class Messaging extends Action Query::equal('$id', $targetIds), Query::limit(\count($targetIds)), ]); + $targets = \array_filter($targets, fn(Document $target) => + $target->getAttribute('providerType') === $message->getAttribute('providerType')); $recipients = \array_merge($recipients, $targets); } From 170160659368a0a8c71cf672d5aa33fe1583b09a Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Fri, 19 Jan 2024 14:41:53 +1300 Subject: [PATCH 28/30] Console warning instead of throw so error reporting isn't triggered --- src/Appwrite/Platform/Workers/Messaging.php | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/Appwrite/Platform/Workers/Messaging.php b/src/Appwrite/Platform/Workers/Messaging.php index 0aa8e53df5..c146412e94 100644 --- a/src/Appwrite/Platform/Workers/Messaging.php +++ b/src/Appwrite/Platform/Workers/Messaging.php @@ -134,7 +134,8 @@ class Messaging extends Action 'deliveryErrors' => ['No valid recipients found.'] ])); - throw new \Exception('No valid recipients found.'); + Console::warning('No valid recipients found.'); + return; } $fallback = $dbForProject->findOne('providers', [ @@ -148,7 +149,8 @@ class Messaging extends Action 'deliveryErrors' => ['No fallback provider found.'] ])); - throw new \Exception('No fallback provider found.'); + Console::warning('No fallback provider found.'); + return; } /** From 65573adad6446c7fbaa1c642ed69106a3cb526bc Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Fri, 19 Jan 2024 16:15:54 +1300 Subject: [PATCH 29/30] Use status enum --- app/controllers/api/messaging.php | 141 ++++++++++++-------- src/Appwrite/Platform/Workers/Messaging.php | 9 +- 2 files changed, 88 insertions(+), 62 deletions(-) diff --git a/app/controllers/api/messaging.php b/app/controllers/api/messaging.php index 8587426b8e..7e27cf15a4 100644 --- a/app/controllers/api/messaging.php +++ b/app/controllers/api/messaging.php @@ -2325,30 +2325,35 @@ App::post('/v1/messaging/messages/email') 'status' => $status, ])); - if ($status === MessageStatus::PROCESSING) { - $queueForMessaging - ->setMessageId($message->getId()) - ->trigger(); - } elseif (!\is_null($scheduledAt)) { - $schedule = $dbForConsole->createDocument('schedules', new Document([ - 'region' => App::getEnv('_APP_REGION', 'default'), - 'resourceType' => 'message', - 'resourceCollection' => 'messages', - 'resourceId' => $message->getId(), - 'resourceInternalId' => $message->getInternalId(), - 'resourceUpdatedAt' => DateTime::now(), - 'projectId' => $project->getId(), - 'schedule' => $message->getAttribute('scheduledAt'), - 'active' => $status === 'processing', - ])); + switch ($status) { + case MessageStatus::PROCESSING: + $queueForMessaging + ->setMessageId($message->getId()) + ->trigger(); + break; + case MessageStatus::SCHEDULED: + $schedule = $dbForConsole->createDocument('schedules', new Document([ + 'region' => App::getEnv('_APP_REGION', 'default'), + 'resourceType' => 'message', + 'resourceCollection' => 'messages', + 'resourceId' => $message->getId(), + 'resourceInternalId' => $message->getInternalId(), + 'resourceUpdatedAt' => DateTime::now(), + 'projectId' => $project->getId(), + 'schedule' => $scheduledAt, + 'active' => true, + ])); - $message->setAttribute('scheduleId', $schedule->getId()); + $message->setAttribute('scheduleId', $schedule->getId()); - $dbForProject->updateDocument( - 'messages', - $message->getId(), - $message - ); + $dbForProject->updateDocument( + 'messages', + $message->getId(), + $message + ); + break; + default: + break; } $queueForEvents @@ -2427,25 +2432,35 @@ App::post('/v1/messaging/messages/sms') 'status' => $status, ])); - if ($status === MessageStatus::PROCESSING) { - $queueForMessaging - ->setMessageId($message->getId()) - ->trigger(); - } elseif ($status === 'processing' && $scheduledAt !== null) { - $schedule = $dbForConsole->createDocument('schedules', new Document([ - 'region' => App::getEnv('_APP_REGION', 'default'), - 'resourceType' => 'message', - 'resourceCollection' => 'messages', - 'resourceId' => $message->getId(), - 'resourceInternalId' => $message->getInternalId(), - 'resourceUpdatedAt' => DateTime::now(), - 'projectId' => $project->getId(), - 'schedule' => $message->getAttribute('scheduledAt'), - 'active' => $status === 'processing', - ])); + switch ($status) { + case MessageStatus::PROCESSING: + $queueForMessaging + ->setMessageId($message->getId()) + ->trigger(); + break; + case MessageStatus::SCHEDULED: + $schedule = $dbForConsole->createDocument('schedules', new Document([ + 'region' => App::getEnv('_APP_REGION', 'default'), + 'resourceType' => 'message', + 'resourceCollection' => 'messages', + 'resourceId' => $message->getId(), + 'resourceInternalId' => $message->getInternalId(), + 'resourceUpdatedAt' => DateTime::now(), + 'projectId' => $project->getId(), + 'schedule' => $scheduledAt, + 'active' => true, + ])); - $message->setAttribute('scheduleId', $schedule->getId()); - $dbForProject->updateDocument('messages', $message->getId(), $message); + $message->setAttribute('scheduleId', $schedule->getId()); + + $dbForProject->updateDocument( + 'messages', + $message->getId(), + $message + ); + break; + default: + break; } $queueForEvents @@ -2541,25 +2556,35 @@ App::post('/v1/messaging/messages/push') 'status' => $status, ])); - if ($status === MessageStatus::PROCESSING) { - $queueForMessaging - ->setMessageId($message->getId()) - ->trigger(); - } elseif ($status === 'processing' && $scheduledAt !== null) { - $schedule = $dbForConsole->createDocument('schedules', new Document([ - 'region' => App::getEnv('_APP_REGION', 'default'), - 'resourceType' => 'message', - 'resourceCollection' => 'messages', - 'resourceId' => $message->getId(), - 'resourceInternalId' => $message->getInternalId(), - 'resourceUpdatedAt' => DateTime::now(), - 'projectId' => $project->getId(), - 'schedule' => $message->getAttribute('scheduledAt'), - 'active' => $status === 'processing', - ])); + switch ($status) { + case MessageStatus::PROCESSING: + $queueForMessaging + ->setMessageId($message->getId()) + ->trigger(); + break; + case MessageStatus::SCHEDULED: + $schedule = $dbForConsole->createDocument('schedules', new Document([ + 'region' => App::getEnv('_APP_REGION', 'default'), + 'resourceType' => 'message', + 'resourceCollection' => 'messages', + 'resourceId' => $message->getId(), + 'resourceInternalId' => $message->getInternalId(), + 'resourceUpdatedAt' => DateTime::now(), + 'projectId' => $project->getId(), + 'schedule' => $scheduledAt, + 'active' => true, + ])); - $message->setAttribute('scheduleId', $schedule->getId()); - $dbForProject->updateDocument('messages', $message->getId(), $message); + $message->setAttribute('scheduleId', $schedule->getId()); + + $dbForProject->updateDocument( + 'messages', + $message->getId(), + $message + ); + break; + default: + break; } $queueForEvents diff --git a/src/Appwrite/Platform/Workers/Messaging.php b/src/Appwrite/Platform/Workers/Messaging.php index c146412e94..5962755317 100644 --- a/src/Appwrite/Platform/Workers/Messaging.php +++ b/src/Appwrite/Platform/Workers/Messaging.php @@ -2,6 +2,7 @@ namespace Appwrite\Platform\Workers; +use Appwrite\Enum\MessageStatus; use Appwrite\Extend\Exception; use Utopia\App; use Utopia\CLI\Console; @@ -130,7 +131,7 @@ class Messaging extends Action if (empty($recipients)) { $dbForProject->updateDocument('messages', $message->getId(), $message->setAttributes([ - 'status' => 'failed', + 'status' => MessageStatus::FAILED, 'deliveryErrors' => ['No valid recipients found.'] ])); @@ -145,7 +146,7 @@ class Messaging extends Action if ($fallback === false || $fallback->isEmpty()) { $dbForProject->updateDocument('messages', $message->getId(), $message->setAttributes([ - 'status' => 'failed', + 'status' => MessageStatus::FAILED, 'deliveryErrors' => ['No fallback provider found.'] ])); @@ -279,9 +280,9 @@ class Messaging extends Action $message->setAttribute('deliveryErrors', $deliveryErrors); if (\count($message->getAttribute('deliveryErrors')) > 0) { - $message->setAttribute('status', 'failed'); + $message->setAttribute('status', MessageStatus::FAILED); } else { - $message->setAttribute('status', 'sent'); + $message->setAttribute('status', MessageStatus::SENT); } $message->removeAttribute('to'); From d292562090963a4627c844cd7e6513001bce4dd2 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Fri, 19 Jan 2024 16:16:24 +1300 Subject: [PATCH 30/30] Throw if status scheduled and no schedule set --- app/config/errors.php | 7 ++++++- app/controllers/api/messaging.php | 12 ++++++++++++ src/Appwrite/Extend/Exception.php | 1 + 3 files changed, 19 insertions(+), 1 deletion(-) diff --git a/app/config/errors.php b/app/config/errors.php index bc2c436313..e5c2d4e5bd 100644 --- a/app/config/errors.php +++ b/app/config/errors.php @@ -4,6 +4,7 @@ * List of server wide error codes and their respective messages. */ +use Appwrite\Enum\MessageStatus; use Appwrite\Extend\Exception; return [ @@ -871,10 +872,14 @@ return [ 'description' => 'Message with the target ID is not a push target.', 'code' => 400, ], + Exception::MESSAGE_MISSING_SCHEDULE => [ + 'name' => Exception::MESSAGE_MISSING_SCHEDULE, + 'description' => 'Message can not have status ' . MessageStatus::SCHEDULED . ' without a schedule.', + 'code' => 400, + ], Exception::SCHEDULE_NOT_FOUND => [ 'name' => Exception::SCHEDULE_NOT_FOUND, 'description' => 'Schedule with the requested ID could not be found.', 'code' => 404, ], - ]; diff --git a/app/controllers/api/messaging.php b/app/controllers/api/messaging.php index 7e27cf15a4..de20c053e0 100644 --- a/app/controllers/api/messaging.php +++ b/app/controllers/api/messaging.php @@ -2287,6 +2287,10 @@ App::post('/v1/messaging/messages/email') throw new Exception(Exception::MESSAGE_MISSING_TARGET); } + if ($status === MessageStatus::SCHEDULED && \is_null($scheduledAt)) { + throw new Exception(Exception::MESSAGE_MISSING_SCHEDULE); + } + $mergedTargets = \array_merge($targets, $cc, $bcc); if (!empty($mergedTargets)) { @@ -2401,6 +2405,10 @@ App::post('/v1/messaging/messages/sms') throw new Exception(Exception::MESSAGE_MISSING_TARGET); } + if ($status === MessageStatus::SCHEDULED && \is_null($scheduledAt)) { + throw new Exception(Exception::MESSAGE_MISSING_SCHEDULE); + } + if (!empty($targets)) { $foundTargets = $dbForProject->find('targets', [ Query::equal('$id', $targets), @@ -2516,6 +2524,10 @@ App::post('/v1/messaging/messages/push') throw new Exception(Exception::MESSAGE_MISSING_TARGET); } + if ($status === MessageStatus::SCHEDULED && \is_null($scheduledAt)) { + throw new Exception(Exception::MESSAGE_MISSING_SCHEDULE); + } + if (!empty($targets)) { $foundTargets = $dbForProject->find('targets', [ Query::equal('$id', $targets), diff --git a/src/Appwrite/Extend/Exception.php b/src/Appwrite/Extend/Exception.php index b1d654c400..183f4a812a 100644 --- a/src/Appwrite/Extend/Exception.php +++ b/src/Appwrite/Extend/Exception.php @@ -262,6 +262,7 @@ class Exception extends \Exception public const MESSAGE_TARGET_NOT_EMAIL = 'message_target_not_email'; public const MESSAGE_TARGET_NOT_SMS = 'message_target_not_sms'; public const MESSAGE_TARGET_NOT_PUSH = 'message_target_not_push'; + public const MESSAGE_MISSING_SCHEDULE = 'message_missing_schedule'; /** Schedules */ public const SCHEDULE_NOT_FOUND = 'schedule_not_found';