From 34075322d735deeef5ce0df15af99941470e42d5 Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Fri, 8 May 2026 14:32:11 +0530 Subject: [PATCH 1/3] Migrate mails and messaging queues to publishers --- app/controllers/api/account.php | 251 ++++---- app/controllers/api/messaging.php | 75 ++- app/controllers/shared/api.php | 16 +- app/init/resources.php | 10 + app/init/resources/request.php | 8 - app/init/worker/message.php | 10 - src/Appwrite/Bus/Listeners/Mails.php | 58 +- src/Appwrite/Event/Mail.php | 576 ------------------ src/Appwrite/Event/Message/Mail.php | 66 ++ src/Appwrite/Event/Message/Messaging.php | 45 ++ src/Appwrite/Event/Messaging.php | 182 ------ src/Appwrite/Event/Publisher/Mail.php | 27 + src/Appwrite/Event/Publisher/Messaging.php | 27 + .../Http/Account/MFA/Challenges/Create.php | 78 +-- .../Health/Http/Health/Queue/Failed/Get.php | 16 +- .../Health/Http/Health/Queue/Mails/Get.php | 8 +- .../Http/Health/Queue/Messaging/Get.php | 8 +- .../Http/Project/SMTP/Tests/Create.php | 42 +- .../Modules/Teams/Http/Memberships/Create.php | 66 +- .../Platform/Tasks/ScheduleMessages.php | 23 +- .../Platform/Workers/Certificates.php | 39 +- src/Appwrite/Platform/Workers/Migrations.php | 48 +- src/Appwrite/Platform/Workers/Webhooks.php | 41 +- 23 files changed, 569 insertions(+), 1151 deletions(-) delete mode 100644 src/Appwrite/Event/Mail.php create mode 100644 src/Appwrite/Event/Message/Mail.php create mode 100644 src/Appwrite/Event/Message/Messaging.php delete mode 100644 src/Appwrite/Event/Messaging.php create mode 100644 src/Appwrite/Event/Publisher/Mail.php create mode 100644 src/Appwrite/Event/Publisher/Messaging.php diff --git a/app/controllers/api/account.php b/app/controllers/api/account.php index ac07efc72b..e01c27e45c 100644 --- a/app/controllers/api/account.php +++ b/app/controllers/api/account.php @@ -13,8 +13,10 @@ use Appwrite\Bus\Events\SessionCreated; use Appwrite\Detector\Detector; use Appwrite\Event\Delete; use Appwrite\Event\Event; -use Appwrite\Event\Mail; -use Appwrite\Event\Messaging; +use Appwrite\Event\Message\Mail as MailMessage; +use Appwrite\Event\Message\Messaging as MessagingMessage; +use Appwrite\Event\Publisher\Mail as MailPublisher; +use Appwrite\Event\Publisher\Messaging as MessagingPublisher; use Appwrite\Extend\Exception; use Appwrite\Hooks\Hooks; use Appwrite\Network\Validator\Redirect; @@ -2113,12 +2115,12 @@ Http::post('/v1/account/tokens/magic-url') ->inject('dbForProject') ->inject('locale') ->inject('queueForEvents') - ->inject('queueForMails') + ->inject('publisherForMails') ->inject('plan') ->inject('proofForPassword') ->inject('platform') ->inject('authorization') - ->action(function (string $userId, string $email, string $url, bool $phrase, Request $request, Response $response, Document $user, Document $project, Database $dbForProject, Locale $locale, Event $queueForEvents, Mail $queueForMails, array $plan, ProofsPassword $proofForPassword, array $platform, Authorization $authorization) { + ->action(function (string $userId, string $email, string $url, bool $phrase, Request $request, Response $response, Document $user, Document $project, Database $dbForProject, Locale $locale, Event $queueForEvents, MailPublisher $publisherForMails, array $plan, ProofsPassword $proofForPassword, array $platform, Authorization $authorization) { if (empty(System::getEnv('_APP_SMTP_HOST'))) { throw new Exception(Exception::GENERAL_SMTP_DISABLED, 'SMTP disabled'); } @@ -2304,6 +2306,7 @@ Http::post('/v1/account/tokens/magic-url') $senderName = System::getEnv('_APP_SYSTEM_EMAIL_NAME', APP_NAME . ' Server'); $replyToEmail = ''; $replyToName = ''; + $smtpConfig = []; if ($smtpEnabled) { if (!empty($smtp['senderEmail'])) { @@ -2321,13 +2324,6 @@ Http::post('/v1/account/tokens/magic-url') $replyToName = $smtp['replyToName']; } - $queueForMails - ->setSmtpHost($smtp['host'] ?? '') - ->setSmtpPort($smtp['port'] ?? '') - ->setSmtpUsername($smtp['username'] ?? '') - ->setSmtpPassword($smtp['password'] ?? '') - ->setSmtpSecure($smtp['secure'] ?? ''); - if (!empty($customTemplate)) { if (!empty($customTemplate['senderEmail'])) { $senderEmail = $customTemplate['senderEmail']; @@ -2348,11 +2344,17 @@ Http::post('/v1/account/tokens/magic-url') $subject = $customTemplate['subject'] ?? $subject; } - $queueForMails - ->setSmtpReplyToEmail($replyToEmail) - ->setSmtpReplyToName($replyToName) - ->setSmtpSenderEmail($senderEmail) - ->setSmtpSenderName($senderName); + $smtpConfig = [ + 'host' => $smtp['host'] ?? '', + 'port' => $smtp['port'] ?? '', + 'username' => $smtp['username'] ?? '', + 'password' => $smtp['password'] ?? '', + 'secure' => $smtp['secure'] ?? '', + 'replyToEmail' => $replyToEmail, + 'replyToName' => $replyToName, + 'senderEmail' => $senderEmail, + 'senderName' => $senderName, + ]; } $projectName = $project->getAttribute('name'); @@ -2374,18 +2376,17 @@ Http::post('/v1/account/tokens/magic-url') 'team' => '', ]; - $queueForMails - ->setSubject($subject) - ->setPreview($preview) - ->setBody($body) - ->appendVariables($emailVariables) - ->setRecipient($email); - - if ($project->getId() === 'console') { - $queueForMails->setSenderName($platform['emailSenderName']); - } - - $queueForMails->trigger(); + $publisherForMails->enqueue(new MailMessage( + project: $project, + recipient: $email, + subject: $subject, + body: $body, + preview: $preview, + smtp: $smtpConfig, + variables: $emailVariables, + customMailOptions: $project->getId() === 'console' ? ['senderName' => $platform['emailSenderName']] : [], + platform: $platform, + )); $token->setAttribute('secret', $tokenSecret); @@ -2436,12 +2437,12 @@ Http::post('/v1/account/tokens/email') ->inject('dbForProject') ->inject('locale') ->inject('queueForEvents') - ->inject('queueForMails') + ->inject('publisherForMails') ->inject('plan') ->inject('proofForPassword') ->inject('proofForCode') ->inject('authorization') - ->action(function (string $userId, string $email, bool $phrase, Request $request, Response $response, User $user, Document $project, array $platform, Database $dbForProject, Locale $locale, Event $queueForEvents, Mail $queueForMails, array $plan, ProofsPassword $proofForPassword, ProofsCode $proofForCode, Authorization $authorization) { + ->action(function (string $userId, string $email, bool $phrase, Request $request, Response $response, User $user, Document $project, array $platform, Database $dbForProject, Locale $locale, Event $queueForEvents, MailPublisher $publisherForMails, array $plan, ProofsPassword $proofForPassword, ProofsCode $proofForCode, Authorization $authorization) { if (empty(System::getEnv('_APP_SMTP_HOST'))) { throw new Exception(Exception::GENERAL_SMTP_DISABLED, 'SMTP disabled'); } @@ -2633,6 +2634,7 @@ Http::post('/v1/account/tokens/email') $senderName = System::getEnv('_APP_SYSTEM_EMAIL_NAME', APP_NAME . ' Server'); $replyToEmail = ''; $replyToName = ''; + $smtpConfig = []; if ($smtpEnabled) { if (!empty($smtp['senderEmail'])) { @@ -2650,13 +2652,6 @@ Http::post('/v1/account/tokens/email') $replyToName = $smtp['replyToName']; } - $queueForMails - ->setSmtpHost($smtp['host'] ?? '') - ->setSmtpPort($smtp['port'] ?? '') - ->setSmtpUsername($smtp['username'] ?? '') - ->setSmtpPassword($smtp['password'] ?? '') - ->setSmtpSecure($smtp['secure'] ?? ''); - if (!empty($customTemplate)) { if (!empty($customTemplate['senderEmail'])) { $senderEmail = $customTemplate['senderEmail']; @@ -2677,11 +2672,17 @@ Http::post('/v1/account/tokens/email') $subject = $customTemplate['subject'] ?? $subject; } - $queueForMails - ->setSmtpReplyToEmail($replyToEmail) - ->setSmtpReplyToName($replyToName) - ->setSmtpSenderEmail($senderEmail) - ->setSmtpSenderName($senderName); + $smtpConfig = [ + 'host' => $smtp['host'] ?? '', + 'port' => $smtp['port'] ?? '', + 'username' => $smtp['username'] ?? '', + 'password' => $smtp['password'] ?? '', + 'secure' => $smtp['secure'] ?? '', + 'replyToEmail' => $replyToEmail, + 'replyToName' => $replyToName, + 'senderEmail' => $senderEmail, + 'senderName' => $senderName, + ]; } $projectName = $project->getAttribute('name'); @@ -2717,20 +2718,18 @@ Http::post('/v1/account/tokens/email') ]); } - $queueForMails - ->setSubject($subject) - ->setPreview($preview) - ->setBody($body) - ->setBodyTemplate($bodyTemplate) - ->appendVariables($emailVariables) - ->setRecipient($email); - - // since this is console project, set email sender name! - if ($smtpBaseTemplate === APP_BRANDED_EMAIL_BASE_TEMPLATE) { - $queueForMails->setSenderName($platform['emailSenderName']); - } - - $queueForMails->trigger(); + $publisherForMails->enqueue(new MailMessage( + project: $project, + recipient: $email, + subject: $subject, + bodyTemplate: $bodyTemplate, + body: $body, + preview: $preview, + smtp: $smtpConfig, + variables: $emailVariables, + customMailOptions: $smtpBaseTemplate === APP_BRANDED_EMAIL_BASE_TEMPLATE ? ['senderName' => $platform['emailSenderName']] : [], + platform: $platform, + )); $token->setAttribute('secret', $tokenSecret); @@ -2880,7 +2879,7 @@ Http::post('/v1/account/tokens/phone') ->inject('platform') ->inject('dbForProject') ->inject('queueForEvents') - ->inject('queueForMessaging') + ->inject('publisherForMessaging') ->inject('locale') ->inject('timelimit') ->inject('usage') @@ -2888,7 +2887,7 @@ Http::post('/v1/account/tokens/phone') ->inject('store') ->inject('proofForCode') ->inject('authorization') - ->action(function (string $userId, string $phone, Request $request, Response $response, User $user, Document $project, array $platform, Database $dbForProject, Event $queueForEvents, Messaging $queueForMessaging, Locale $locale, callable $timelimit, Context $usage, array $plan, Store $store, ProofsCode $proofForCode, Authorization $authorization) { + ->action(function (string $userId, string $phone, Request $request, Response $response, User $user, Document $project, array $platform, Database $dbForProject, Event $queueForEvents, MessagingPublisher $publisherForMessaging, Locale $locale, callable $timelimit, Context $usage, array $plan, Store $store, ProofsCode $proofForCode, Authorization $authorization) { if (empty(System::getEnv('_APP_SMS_PROVIDER'))) { throw new Exception(Exception::GENERAL_PHONE_DISABLED, 'Phone provider not configured'); } @@ -3021,11 +3020,13 @@ Http::post('/v1/account/tokens/phone') ], ]); - $queueForMessaging - ->setType(MESSAGE_SEND_TYPE_INTERNAL) - ->setMessage($messageDoc) - ->setRecipients([$phone]) - ->setProviderType(MESSAGE_TYPE_SMS); + $publisherForMessaging->enqueue(new MessagingMessage( + type: MESSAGE_SEND_TYPE_INTERNAL, + project: $project, + message: $messageDoc, + recipients: [$phone], + providerType: MESSAGE_TYPE_SMS, + )); $helper = PhoneNumberUtil::getInstance(); try { @@ -3681,11 +3682,11 @@ Http::post('/v1/account/recovery') ->inject('project') ->inject('platform') ->inject('locale') - ->inject('queueForMails') + ->inject('publisherForMails') ->inject('queueForEvents') ->inject('proofForToken') ->inject('authorization') - ->action(function (string $email, string $url, Request $request, Response $response, User $user, Database $dbForProject, Document $project, array $platform, Locale $locale, Mail $queueForMails, Event $queueForEvents, ProofsToken $proofForToken, Authorization $authorization) { + ->action(function (string $email, string $url, Request $request, Response $response, User $user, Database $dbForProject, Document $project, array $platform, Locale $locale, MailPublisher $publisherForMails, Event $queueForEvents, ProofsToken $proofForToken, Authorization $authorization) { if (empty(System::getEnv('_APP_SMTP_HOST'))) { throw new Exception(Exception::GENERAL_SMTP_DISABLED, 'SMTP Disabled'); @@ -3768,6 +3769,7 @@ Http::post('/v1/account/recovery') $senderName = System::getEnv('_APP_SYSTEM_EMAIL_NAME', APP_NAME . ' Server'); $replyToEmail = ''; $replyToName = ''; + $smtpConfig = []; if ($smtpEnabled) { if (!empty($smtp['senderEmail'])) { @@ -3785,13 +3787,6 @@ Http::post('/v1/account/recovery') $replyToName = $smtp['replyToName']; } - $queueForMails - ->setSmtpHost($smtp['host'] ?? '') - ->setSmtpPort($smtp['port'] ?? '') - ->setSmtpUsername($smtp['username'] ?? '') - ->setSmtpPassword($smtp['password'] ?? '') - ->setSmtpSecure($smtp['secure'] ?? ''); - if (!empty($customTemplate)) { if (!empty($customTemplate['senderEmail'])) { $senderEmail = $customTemplate['senderEmail']; @@ -3812,11 +3807,17 @@ Http::post('/v1/account/recovery') $subject = $customTemplate['subject'] ?? $subject; } - $queueForMails - ->setSmtpReplyToEmail($replyToEmail) - ->setSmtpReplyToName($replyToName) - ->setSmtpSenderEmail($senderEmail) - ->setSmtpSenderName($senderName); + $smtpConfig = [ + 'host' => $smtp['host'] ?? '', + 'port' => $smtp['port'] ?? '', + 'username' => $smtp['username'] ?? '', + 'password' => $smtp['password'] ?? '', + 'secure' => $smtp['secure'] ?? '', + 'replyToEmail' => $replyToEmail, + 'replyToName' => $replyToName, + 'senderEmail' => $senderEmail, + 'senderName' => $senderName, + ]; } $emailVariables = [ @@ -3829,19 +3830,18 @@ Http::post('/v1/account/recovery') 'team' => '' ]; - $queueForMails - ->setRecipient($profile->getAttribute('email', '')) - ->setName($profile->getAttribute('name', '')) - ->setBody($body) - ->appendVariables($emailVariables) - ->setSubject($subject) - ->setPreview($preview); - - if ($project->getId() === 'console') { - $queueForMails->setSenderName($platform['emailSenderName']); - } - - $queueForMails->trigger(); + $publisherForMails->enqueue(new MailMessage( + project: $project, + recipient: $profile->getAttribute('email', ''), + name: $profile->getAttribute('name', ''), + subject: $subject, + body: $body, + preview: $preview, + smtp: $smtpConfig, + variables: $emailVariables, + customMailOptions: $project->getId() === 'console' ? ['senderName' => $platform['emailSenderName']] : [], + platform: $platform, + )); $recovery->setAttribute('secret', $secret); @@ -4009,10 +4009,10 @@ Http::post('/v1/account/verifications/email') ->inject('dbForProject') ->inject('locale') ->inject('queueForEvents') - ->inject('queueForMails') + ->inject('publisherForMails') ->inject('proofForToken') ->inject('authorization') - ->action(function (string $url, Request $request, Response $response, Document $project, array $platform, User $user, Database $dbForProject, Locale $locale, Event $queueForEvents, Mail $queueForMails, ProofsToken $proofForToken, Authorization $authorization) { + ->action(function (string $url, Request $request, Response $response, Document $project, array $platform, User $user, Database $dbForProject, Locale $locale, Event $queueForEvents, MailPublisher $publisherForMails, ProofsToken $proofForToken, Authorization $authorization) { if (empty(System::getEnv('_APP_SMTP_HOST'))) { throw new Exception(Exception::GENERAL_SMTP_DISABLED, 'SMTP Disabled'); @@ -4099,6 +4099,7 @@ Http::post('/v1/account/verifications/email') $senderName = System::getEnv('_APP_SYSTEM_EMAIL_NAME', APP_NAME . ' Server'); $replyToEmail = ''; $replyToName = ''; + $smtpConfig = []; if ($smtpEnabled) { if (!empty($smtp['senderEmail'])) { @@ -4116,13 +4117,6 @@ Http::post('/v1/account/verifications/email') $replyToName = $smtp['replyToName']; } - $queueForMails - ->setSmtpHost($smtp['host'] ?? '') - ->setSmtpPort($smtp['port'] ?? '') - ->setSmtpUsername($smtp['username'] ?? '') - ->setSmtpPassword($smtp['password'] ?? '') - ->setSmtpSecure($smtp['secure'] ?? ''); - if (!empty($customTemplate)) { if (!empty($customTemplate['senderEmail'])) { $senderEmail = $customTemplate['senderEmail']; @@ -4143,11 +4137,17 @@ Http::post('/v1/account/verifications/email') $subject = $customTemplate['subject'] ?? $subject; } - $queueForMails - ->setSmtpReplyToEmail($replyToEmail) - ->setSmtpReplyToName($replyToName) - ->setSmtpSenderEmail($senderEmail) - ->setSmtpSenderName($senderName); + $smtpConfig = [ + 'host' => $smtp['host'] ?? '', + 'port' => $smtp['port'] ?? '', + 'username' => $smtp['username'] ?? '', + 'password' => $smtp['password'] ?? '', + 'secure' => $smtp['secure'] ?? '', + 'replyToEmail' => $replyToEmail, + 'replyToName' => $replyToName, + 'senderEmail' => $senderEmail, + 'senderName' => $senderName, + ]; } $emailVariables = [ @@ -4174,20 +4174,19 @@ Http::post('/v1/account/verifications/email') ]); } - $queueForMails - ->setSubject($subject) - ->setPreview($preview) - ->setBody($body) - ->setBodyTemplate($bodyTemplate) - ->appendVariables($emailVariables) - ->setRecipient($user->getAttribute('email')) - ->setName($user->getAttribute('name') ?? ''); - - if ($project->getId() === 'console') { - $queueForMails->setSenderName($platform['emailSenderName']); - } - - $queueForMails->trigger(); + $publisherForMails->enqueue(new MailMessage( + project: $project, + recipient: $user->getAttribute('email'), + name: $user->getAttribute('name') ?? '', + subject: $subject, + bodyTemplate: $bodyTemplate, + body: $body, + preview: $preview, + smtp: $smtpConfig, + variables: $emailVariables, + customMailOptions: $project->getId() === 'console' ? ['senderName' => $platform['emailSenderName']] : [], + platform: $platform, + )); $verification->setAttribute('secret', $verificationSecret); @@ -4321,7 +4320,7 @@ Http::post('/v1/account/verifications/phone') ->inject('user') ->inject('dbForProject') ->inject('queueForEvents') - ->inject('queueForMessaging') + ->inject('publisherForMessaging') ->inject('project') ->inject('locale') ->inject('timelimit') @@ -4329,7 +4328,7 @@ Http::post('/v1/account/verifications/phone') ->inject('plan') ->inject('proofForCode') ->inject('authorization') - ->action(function (Request $request, Response $response, User $user, Database $dbForProject, Event $queueForEvents, Messaging $queueForMessaging, Document $project, Locale $locale, callable $timelimit, Context $usage, array $plan, ProofsCode $proofForCode, Authorization $authorization) { + ->action(function (Request $request, Response $response, User $user, Database $dbForProject, Event $queueForEvents, MessagingPublisher $publisherForMessaging, Document $project, Locale $locale, callable $timelimit, Context $usage, array $plan, ProofsCode $proofForCode, Authorization $authorization) { if (empty(System::getEnv('_APP_SMS_PROVIDER'))) { throw new Exception(Exception::GENERAL_PHONE_DISABLED, 'Phone provider not configured'); } @@ -4398,11 +4397,13 @@ Http::post('/v1/account/verifications/phone') ], ]); - $queueForMessaging - ->setType(MESSAGE_SEND_TYPE_INTERNAL) - ->setMessage($messageDoc) - ->setRecipients([$user->getAttribute('phone')]) - ->setProviderType(MESSAGE_TYPE_SMS); + $publisherForMessaging->enqueue(new MessagingMessage( + type: MESSAGE_SEND_TYPE_INTERNAL, + project: $project, + message: $messageDoc, + recipients: [$user->getAttribute('phone')], + providerType: MESSAGE_TYPE_SMS, + )); $helper = PhoneNumberUtil::getInstance(); try { diff --git a/app/controllers/api/messaging.php b/app/controllers/api/messaging.php index 58c6a2c29e..f59f606174 100644 --- a/app/controllers/api/messaging.php +++ b/app/controllers/api/messaging.php @@ -5,7 +5,8 @@ use Appwrite\Auth\Validator\Phone; use Appwrite\Detector\Detector; use Appwrite\Event\Delete; use Appwrite\Event\Event; -use Appwrite\Event\Messaging; +use Appwrite\Event\Message\Messaging as MessagingMessage; +use Appwrite\Event\Publisher\Messaging as MessagingPublisher; use Appwrite\Extend\Exception; use Appwrite\Messaging\Status as MessageStatus; use Appwrite\Permission; @@ -3187,9 +3188,9 @@ Http::post('/v1/messaging/messages/email') ->inject('dbForProject') ->inject('dbForPlatform') ->inject('project') - ->inject('queueForMessaging') + ->inject('publisherForMessaging') ->inject('response') - ->action(function (string $messageId, string $subject, string $content, ?array $topics, ?array $users, ?array $targets, ?array $cc, ?array $bcc, ?array $attachments, bool $draft, bool $html, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Database $dbForPlatform, Document $project, Messaging $queueForMessaging, Response $response) { + ->action(function (string $messageId, string $subject, string $content, ?array $topics, ?array $users, ?array $targets, ?array $cc, ?array $bcc, ?array $attachments, bool $draft, bool $html, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Database $dbForPlatform, Document $project, MessagingPublisher $publisherForMessaging, Response $response) { $messageId = $messageId == 'unique()' ? ID::unique() : $messageId; @@ -3274,9 +3275,11 @@ Http::post('/v1/messaging/messages/email') switch ($status) { case MessageStatus::PROCESSING: - $queueForMessaging - ->setType(MESSAGE_SEND_TYPE_EXTERNAL) - ->setMessageId($message->getId()); + $publisherForMessaging->enqueue(new MessagingMessage( + type: MESSAGE_SEND_TYPE_EXTERNAL, + project: $project, + messageId: $message->getId(), + )); break; case MessageStatus::SCHEDULED: $schedule = $dbForPlatform->createDocument('schedules', new Document([ @@ -3362,9 +3365,9 @@ Http::post('/v1/messaging/messages/sms') ->inject('dbForProject') ->inject('dbForPlatform') ->inject('project') - ->inject('queueForMessaging') + ->inject('publisherForMessaging') ->inject('response') - ->action(function (string $messageId, string $content, ?array $topics, ?array $users, ?array $targets, bool $draft, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Database $dbForPlatform, Document $project, Messaging $queueForMessaging, Response $response) { + ->action(function (string $messageId, string $content, ?array $topics, ?array $users, ?array $targets, bool $draft, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Database $dbForPlatform, Document $project, MessagingPublisher $publisherForMessaging, Response $response) { $messageId = $messageId == 'unique()' ? ID::unique() : $messageId; @@ -3418,9 +3421,11 @@ Http::post('/v1/messaging/messages/sms') switch ($status) { case MessageStatus::PROCESSING: - $queueForMessaging - ->setType(MESSAGE_SEND_TYPE_EXTERNAL) - ->setMessageId($message->getId()); + $publisherForMessaging->enqueue(new MessagingMessage( + type: MESSAGE_SEND_TYPE_EXTERNAL, + project: $project, + messageId: $message->getId(), + )); break; case MessageStatus::SCHEDULED: $schedule = $dbForPlatform->createDocument('schedules', new Document([ @@ -3498,10 +3503,10 @@ Http::post('/v1/messaging/messages/push') ->inject('dbForProject') ->inject('dbForPlatform') ->inject('project') - ->inject('queueForMessaging') + ->inject('publisherForMessaging') ->inject('response') ->inject('platform') - ->action(function (string $messageId, string $title, string $body, ?array $topics, ?array $users, ?array $targets, ?array $data, string $action, string $image, string $icon, string $sound, string $color, string $tag, int $badge, bool $draft, ?string $scheduledAt, bool $contentAvailable, bool $critical, string $priority, Event $queueForEvents, Database $dbForProject, Database $dbForPlatform, Document $project, Messaging $queueForMessaging, Response $response, array $platform) { + ->action(function (string $messageId, string $title, string $body, ?array $topics, ?array $users, ?array $targets, ?array $data, string $action, string $image, string $icon, string $sound, string $color, string $tag, int $badge, bool $draft, ?string $scheduledAt, bool $contentAvailable, bool $critical, string $priority, Event $queueForEvents, Database $dbForProject, Database $dbForPlatform, Document $project, MessagingPublisher $publisherForMessaging, Response $response, array $platform) { $messageId = $messageId == 'unique()' ? ID::unique() : $messageId; @@ -3638,9 +3643,11 @@ Http::post('/v1/messaging/messages/push') switch ($status) { case MessageStatus::PROCESSING: - $queueForMessaging - ->setType(MESSAGE_SEND_TYPE_EXTERNAL) - ->setMessageId($message->getId()); + $publisherForMessaging->enqueue(new MessagingMessage( + type: MESSAGE_SEND_TYPE_EXTERNAL, + project: $project, + messageId: $message->getId(), + )); break; case MessageStatus::SCHEDULED: $schedule = $dbForPlatform->createDocument('schedules', new Document([ @@ -3983,9 +3990,9 @@ Http::patch('/v1/messaging/messages/email/:messageId') ->inject('dbForProject') ->inject('dbForPlatform') ->inject('project') - ->inject('queueForMessaging') + ->inject('publisherForMessaging') ->inject('response') - ->action(function (string $messageId, ?array $topics, ?array $users, ?array $targets, ?string $subject, ?string $content, ?bool $draft, ?bool $html, ?array $cc, ?array $bcc, ?string $scheduledAt, ?array $attachments, Event $queueForEvents, Database $dbForProject, Database $dbForPlatform, Document $project, Messaging $queueForMessaging, Response $response) { + ->action(function (string $messageId, ?array $topics, ?array $users, ?array $targets, ?string $subject, ?string $content, ?bool $draft, ?bool $html, ?array $cc, ?array $bcc, ?string $scheduledAt, ?array $attachments, Event $queueForEvents, Database $dbForProject, Database $dbForPlatform, Document $project, MessagingPublisher $publisherForMessaging, Response $response) { $message = $dbForProject->getDocument('messages', $messageId); if ($message->isEmpty()) { @@ -4141,9 +4148,11 @@ Http::patch('/v1/messaging/messages/email/:messageId') $message = $dbForProject->updateDocument('messages', $message->getId(), $message); if ($status === MessageStatus::PROCESSING) { - $queueForMessaging - ->setType(MESSAGE_SEND_TYPE_EXTERNAL) - ->setMessageId($message->getId()); + $publisherForMessaging->enqueue(new MessagingMessage( + type: MESSAGE_SEND_TYPE_EXTERNAL, + project: $project, + messageId: $message->getId(), + )); } $queueForEvents @@ -4205,9 +4214,9 @@ Http::patch('/v1/messaging/messages/sms/:messageId') ->inject('dbForProject') ->inject('dbForPlatform') ->inject('project') - ->inject('queueForMessaging') + ->inject('publisherForMessaging') ->inject('response') - ->action(function (string $messageId, ?array $topics, ?array $users, ?array $targets, ?string $content, ?bool $draft, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Database $dbForPlatform, Document $project, Messaging $queueForMessaging, Response $response) { + ->action(function (string $messageId, ?array $topics, ?array $users, ?array $targets, ?string $content, ?bool $draft, ?string $scheduledAt, Event $queueForEvents, Database $dbForProject, Database $dbForPlatform, Document $project, MessagingPublisher $publisherForMessaging, Response $response) { $message = $dbForProject->getDocument('messages', $messageId); if ($message->isEmpty()) { @@ -4323,9 +4332,11 @@ Http::patch('/v1/messaging/messages/sms/:messageId') $message = $dbForProject->updateDocument('messages', $message->getId(), $message); if ($status === MessageStatus::PROCESSING) { - $queueForMessaging - ->setType(MESSAGE_SEND_TYPE_EXTERNAL) - ->setMessageId($message->getId()); + $publisherForMessaging->enqueue(new MessagingMessage( + type: MESSAGE_SEND_TYPE_EXTERNAL, + project: $project, + messageId: $message->getId(), + )); } $queueForEvents @@ -4379,10 +4390,10 @@ Http::patch('/v1/messaging/messages/push/:messageId') ->inject('dbForProject') ->inject('dbForPlatform') ->inject('project') - ->inject('queueForMessaging') + ->inject('publisherForMessaging') ->inject('response') ->inject('platform') - ->action(function (string $messageId, ?array $topics, ?array $users, ?array $targets, ?string $title, ?string $body, ?array $data, ?string $action, ?string $image, ?string $icon, ?string $sound, ?string $color, ?string $tag, ?int $badge, ?bool $draft, ?string $scheduledAt, ?bool $contentAvailable, ?bool $critical, ?string $priority, Event $queueForEvents, Database $dbForProject, Database $dbForPlatform, Document $project, Messaging $queueForMessaging, Response $response, array $platform) { + ->action(function (string $messageId, ?array $topics, ?array $users, ?array $targets, ?string $title, ?string $body, ?array $data, ?string $action, ?string $image, ?string $icon, ?string $sound, ?string $color, ?string $tag, ?int $badge, ?bool $draft, ?string $scheduledAt, ?bool $contentAvailable, ?bool $critical, ?string $priority, Event $queueForEvents, Database $dbForProject, Database $dbForPlatform, Document $project, MessagingPublisher $publisherForMessaging, Response $response, array $platform) { $message = $dbForProject->getDocument('messages', $messageId); if ($message->isEmpty()) { @@ -4584,9 +4595,11 @@ Http::patch('/v1/messaging/messages/push/:messageId') $message = $dbForProject->updateDocument('messages', $message->getId(), $message); if ($status === MessageStatus::PROCESSING) { - $queueForMessaging - ->setType(MESSAGE_SEND_TYPE_EXTERNAL) - ->setMessageId($message->getId()); + $publisherForMessaging->enqueue(new MessagingMessage( + type: MESSAGE_SEND_TYPE_EXTERNAL, + project: $project, + messageId: $message->getId(), + )); } $queueForEvents diff --git a/app/controllers/shared/api.php b/app/controllers/shared/api.php index 14ffdc059f..8365274e98 100644 --- a/app/controllers/shared/api.php +++ b/app/controllers/shared/api.php @@ -8,10 +8,8 @@ use Appwrite\Event\Database as EventDatabase; use Appwrite\Event\Delete; use Appwrite\Event\Event; use Appwrite\Event\Func; -use Appwrite\Event\Mail; use Appwrite\Event\Message\Audit as AuditMessage; use Appwrite\Event\Message\Usage as UsageMessage; -use Appwrite\Event\Messaging; use Appwrite\Event\Publisher\Audit; use Appwrite\Event\Publisher\Usage as UsagePublisher; use Appwrite\Event\Realtime; @@ -486,13 +484,11 @@ Http::init() ->inject('project') ->inject('user') ->inject('queueForEvents') - ->inject('queueForMessaging') ->inject('auditContext') ->inject('queueForDeletes') ->inject('queueForDatabase') ->inject('usage') ->inject('queueForFunctions') - ->inject('queueForMails') ->inject('dbForProject') ->inject('timelimit') ->inject('resourceToken') @@ -504,7 +500,7 @@ Http::init() ->inject('platform') ->inject('authorization') ->inject('cacheControlForStorage') - ->action(function (Http $utopia, Request $request, Response $response, Document $project, User $user, Event $queueForEvents, Messaging $queueForMessaging, AuditContext $auditContext, Delete $queueForDeletes, EventDatabase $queueForDatabase, Context $usage, Func $queueForFunctions, Mail $queueForMails, Database $dbForProject, callable $timelimit, Document $resourceToken, string $mode, ?Key $apiKey, array $plan, Document $devKey, Telemetry $telemetry, array $platform, Authorization $authorization, callable $cacheControlForStorage) { + ->action(function (Http $utopia, Request $request, Response $response, Document $project, User $user, Event $queueForEvents, AuditContext $auditContext, Delete $queueForDeletes, EventDatabase $queueForDatabase, Context $usage, Func $queueForFunctions, Database $dbForProject, callable $timelimit, Document $resourceToken, string $mode, ?Key $apiKey, array $plan, Document $devKey, Telemetry $telemetry, array $platform, Authorization $authorization, callable $cacheControlForStorage) { $response->setUser($user); $request->setUser($user); @@ -617,13 +613,10 @@ Http::init() /* Auto-set projects */ $queueForDeletes->setProject($project); $queueForDatabase->setProject($project); - $queueForMessaging->setProject($project); $queueForFunctions->setProject($project); - $queueForMails->setProject($project); /* Auto-set platforms */ $queueForFunctions->setPlatform($platform); - $queueForMails->setPlatform($platform); $useCache = $route->getLabel('cache', false); $storageCacheOperationsCounter = $telemetry->createCounter('storage.cache.operations.load'); @@ -815,7 +808,6 @@ Http::shutdown() ->inject('publisherForUsage') ->inject('queueForDeletes') ->inject('queueForDatabase') - ->inject('queueForMessaging') ->inject('queueForFunctions') ->inject('queueForWebhooks') ->inject('queueForRealtime') @@ -826,7 +818,7 @@ Http::shutdown() ->inject('bus') ->inject('apiKey') ->inject('mode') - ->action(function (Http $utopia, Request $request, Response $response, Document $project, User $user, Event $queueForEvents, AuditContext $auditContext, Audit $publisherForAudits, Context $usage, UsagePublisher $publisherForUsage, Delete $queueForDeletes, EventDatabase $queueForDatabase, Messaging $queueForMessaging, Func $queueForFunctions, Event $queueForWebhooks, Realtime $queueForRealtime, Database $dbForProject, Authorization $authorization, callable $timelimit, EventProcessor $eventProcessor, Bus $bus, ?Key $apiKey, string $mode) use ($parseLabel) { + ->action(function (Http $utopia, Request $request, Response $response, Document $project, User $user, Event $queueForEvents, AuditContext $auditContext, Audit $publisherForAudits, Context $usage, UsagePublisher $publisherForUsage, Delete $queueForDeletes, EventDatabase $queueForDatabase, Func $queueForFunctions, Event $queueForWebhooks, Realtime $queueForRealtime, Database $dbForProject, Authorization $authorization, callable $timelimit, EventProcessor $eventProcessor, Bus $bus, ?Key $apiKey, string $mode) use ($parseLabel) { $responsePayload = $response->getPayload(); @@ -975,10 +967,6 @@ Http::shutdown() $queueForDatabase->trigger(); } - if (! empty($queueForMessaging->getType())) { - $queueForMessaging->trigger(); - } - // Cache label $useCache = $route->getLabel('cache', false); if ($useCache) { diff --git a/app/init/resources.php b/app/init/resources.php index c5d034a125..a626b612cb 100644 --- a/app/init/resources.php +++ b/app/init/resources.php @@ -5,6 +5,8 @@ use Appwrite\Event\Publisher\Audit as AuditPublisher; use Appwrite\Event\Publisher\Build as BuildPublisher; use Appwrite\Event\Publisher\Certificate as CertificatePublisher; use Appwrite\Event\Publisher\Execution as ExecutionPublisher; +use Appwrite\Event\Publisher\Mail as MailPublisher; +use Appwrite\Event\Publisher\Messaging as MessagingPublisher; use Appwrite\Event\Publisher\Migration as MigrationPublisher; use Appwrite\Event\Publisher\Screenshot as ScreenshotPublisher; use Appwrite\Event\Publisher\StatsResources as StatsResourcesPublisher; @@ -118,6 +120,14 @@ $container->set('publisherForBuilds', fn (Publisher $publisher) => new BuildPubl $publisher, new Queue(System::getEnv('_APP_BUILDS_QUEUE_NAME', Event::BUILDS_QUEUE_NAME)) ), ['publisher']); +$container->set('publisherForMails', fn (Publisher $publisher) => new MailPublisher( + $publisher, + new Queue(System::getEnv('_APP_MAILS_QUEUE_NAME', Event::MAILS_QUEUE_NAME)) +), ['publisher']); +$container->set('publisherForMessaging', fn (Publisher $publisher) => new MessagingPublisher( + $publisher, + new Queue(System::getEnv('_APP_MESSAGING_QUEUE_NAME', Event::MESSAGING_QUEUE_NAME)) +), ['publisher']); /** * Platform configuration diff --git a/app/init/resources/request.php b/app/init/resources/request.php index 9fd282c4ed..6ed377d9ae 100644 --- a/app/init/resources/request.php +++ b/app/init/resources/request.php @@ -9,8 +9,6 @@ use Appwrite\Event\Database as EventDatabase; use Appwrite\Event\Delete; use Appwrite\Event\Event; use Appwrite\Event\Func; -use Appwrite\Event\Mail; -use Appwrite\Event\Messaging; use Appwrite\Event\Realtime; use Appwrite\Event\Webhook; use Appwrite\Extend\Exception; @@ -116,12 +114,6 @@ return function (Container $container): void { }); // Per-request queue resources (stateful, accumulate event data during request) - $container->set('queueForMessaging', function (Publisher $publisher) { - return new Messaging($publisher); - }, ['publisher']); - $container->set('queueForMails', function (Publisher $publisher) { - return new Mail($publisher); - }, ['publisher']); $container->set('queueForDatabase', function (Publisher $publisher) { return new EventDatabase($publisher); }, ['publisher']); diff --git a/app/init/worker/message.php b/app/init/worker/message.php index 1469934ad4..791bf5edf0 100644 --- a/app/init/worker/message.php +++ b/app/init/worker/message.php @@ -4,8 +4,6 @@ use Appwrite\Event\Database as EventDatabase; use Appwrite\Event\Delete; use Appwrite\Event\Event; use Appwrite\Event\Func; -use Appwrite\Event\Mail; -use Appwrite\Event\Messaging; use Appwrite\Event\Realtime; use Appwrite\Event\Webhook; use Appwrite\Usage\Context; @@ -333,14 +331,6 @@ return function (Container $container): void { return new EventDatabase($publisher); }, ['publisher']); - $container->set('queueForMessaging', function (Publisher $publisher) { - return new Messaging($publisher); - }, ['publisher']); - - $container->set('queueForMails', function (Publisher $publisher) { - return new Mail($publisher); - }, ['publisher']); - $container->set('queueForDeletes', function (Publisher $publisher) { return new Delete($publisher); }, ['publisher']); diff --git a/src/Appwrite/Bus/Listeners/Mails.php b/src/Appwrite/Bus/Listeners/Mails.php index 9b3d68519f..eb36e0d394 100644 --- a/src/Appwrite/Bus/Listeners/Mails.php +++ b/src/Appwrite/Bus/Listeners/Mails.php @@ -4,14 +4,14 @@ namespace Appwrite\Bus\Listeners; use Appwrite\Auth\MFA\Type; use Appwrite\Bus\Events\SessionCreated; -use Appwrite\Event\Mail; +use Appwrite\Event\Message\Mail as MailMessage; +use Appwrite\Event\Publisher\Mail as MailPublisher; use Appwrite\Template\Template; use Utopia\Bus\Listener; use Utopia\Database\Database; use Utopia\Database\Document; use Utopia\Database\Query; use Utopia\Locale\Locale; -use Utopia\Queue\Publisher; use Utopia\Storage\Validator\FileName; use Utopia\System\System; @@ -31,14 +31,14 @@ class Mails extends Listener { $this ->desc('Sends session alert emails') - ->inject('publisher') + ->inject('publisherForMails') ->inject('locale') ->inject('platform') ->inject('dbForProject') ->callback($this->handle(...)); } - public function handle(SessionCreated $event, Publisher $publisher, Locale $locale, array $platform, Database $dbForProject): void + public function handle(SessionCreated $event, MailPublisher $publisherForMails, Locale $locale, array $platform, Database $dbForProject): void { $project = new Document($event->project); @@ -124,34 +124,32 @@ class Mails extends Listener ]; } - $queueForMails = new Mail($publisher); - + $smtpConfig = []; if ($smtp['enabled'] ?? false) { - $queueForMails - ->setSmtpHost($smtp['host'] ?? '') - ->setSmtpPort($smtp['port'] ?? '') - ->setSmtpUsername($smtp['username'] ?? '') - ->setSmtpPassword($smtp['password'] ?? '') - ->setSmtpSecure($smtp['secure'] ?? '') - ->setSmtpReplyToEmail($customTemplate['replyToEmail'] ?? $customTemplate['replyTo'] ?? $smtp['replyToEmail'] ?? $smtp['replyTo'] ?? '') // Includes backwards compatibility - ->setSmtpReplyToName($customTemplate['replyToName'] ?? $smtp['replyToName'] ?? '') - ->setSmtpSenderEmail($customTemplate['senderEmail'] ?? $smtp['senderEmail'] ?? System::getEnv('_APP_SYSTEM_EMAIL_ADDRESS', APP_EMAIL_TEAM)) - ->setSmtpSenderName($customTemplate['senderName'] ?? $smtp['senderName'] ?? System::getEnv('_APP_SYSTEM_EMAIL_NAME', APP_NAME . ' Server')); + $smtpConfig = [ + 'host' => $smtp['host'] ?? '', + 'port' => $smtp['port'] ?? '', + 'username' => $smtp['username'] ?? '', + 'password' => $smtp['password'] ?? '', + 'secure' => $smtp['secure'] ?? '', + 'replyToEmail' => $customTemplate['replyToEmail'] ?? $customTemplate['replyTo'] ?? $smtp['replyToEmail'] ?? $smtp['replyTo'] ?? '', // Includes backwards compatibility + 'replyToName' => $customTemplate['replyToName'] ?? $smtp['replyToName'] ?? '', + 'senderEmail' => $customTemplate['senderEmail'] ?? $smtp['senderEmail'] ?? System::getEnv('_APP_SYSTEM_EMAIL_ADDRESS', APP_EMAIL_TEAM), + 'senderName' => $customTemplate['senderName'] ?? $smtp['senderName'] ?? System::getEnv('_APP_SYSTEM_EMAIL_NAME', APP_NAME . ' Server'), + ]; } - $queueForMails - ->setProject($project) - ->setSubject($subject) - ->setPreview($preview) - ->setBody($body) - ->setBodyTemplate(__DIR__ . '/../../../../app/config/locale/templates/' . $smtpBaseTemplate . '.tpl') - ->appendVariables($emailVariables) - ->setRecipient($event->user['email']); - - if ($isBranded) { - $queueForMails->setSenderName($platform['emailSenderName']); - } - - $queueForMails->trigger(); + $publisherForMails->enqueue(new MailMessage( + project: $project, + recipient: $event->user['email'], + subject: $subject, + bodyTemplate: __DIR__ . '/../../../../app/config/locale/templates/' . $smtpBaseTemplate . '.tpl', + body: $body, + preview: $preview, + smtp: $smtpConfig, + variables: $emailVariables, + customMailOptions: $isBranded ? ['senderName' => $platform['emailSenderName']] : [], + platform: $platform, + )); } } diff --git a/src/Appwrite/Event/Mail.php b/src/Appwrite/Event/Mail.php deleted file mode 100644 index 0685586c60..0000000000 --- a/src/Appwrite/Event/Mail.php +++ /dev/null @@ -1,576 +0,0 @@ -setQueue(System::getEnv('_APP_MAILS_QUEUE_NAME', Event::MAILS_QUEUE_NAME)) - ->setClass(System::getEnv('_APP_MAILS_CLASS_NAME', Event::MAILS_CLASS_NAME)); - } - - /** - * Sets subject for the mail event. - * - * @param string $subject - * @return self - */ - public function setSubject(string $subject): self - { - $this->subject = $subject; - - return $this; - } - - /** - * Returns subject for the mail event. - * - * @return string - */ - public function getSubject(): string - { - return $this->subject; - } - - /** - * Sets recipient for the mail event. - * - * @param string $recipient - * @return self - */ - public function setRecipient(string $recipient): self - { - $this->recipient = $recipient; - - return $this; - } - - /** - * Returns set recipient for mail event. - * - * @return string - */ - public function getRecipient(): string - { - return $this->recipient; - } - - /** - * Sets body for the mail event. - * - * @param string $body - * @return self - */ - public function setBody(string $body): self - { - $this->body = $body; - - return $this; - } - - /** - * Returns body for the mail event. - * - * @return string - */ - public function getBody(): string - { - return $this->body; - } - - /** - * Sets preview for the mail event. - * - * @return self - */ - public function setPreview(string $preview): self - { - $this->preview = $preview; - - return $this; - } - - /** - * Returns preview for the mail event. - * - * @return string - */ - public function getPreview(): string - { - return $this->preview; - } - - /** - * Sets name for the mail event. - * - * @param string $name - * @return self - */ - public function setName(string $name): self - { - $this->name = $name; - - return $this; - } - - /** - * Returns set name for the mail event. - * - * @return string - */ - public function getName(): string - { - return $this->name; - } - - /** - * Sets bodyTemplate for the mail event. - * - * @param string $bodyTemplate - * @return self - */ - public function setBodyTemplate(string $bodyTemplate): self - { - $this->bodyTemplate = $bodyTemplate; - - return $this; - } - - /** - * Returns subject for the mail event. - * - * @return string - */ - public function getBodyTemplate(): string - { - return $this->bodyTemplate; - } - - /** - * Set SMTP Host - * - * @param string $host - * @return self - */ - public function setSmtpHost(string $host): self - { - $this->smtp['host'] = $host; - return $this; - } - - /** - * Set SMTP port - * - * @param int $port - * @return self - */ - public function setSmtpPort(int $port): self - { - $this->smtp['port'] = $port; - return $this; - } - - /** - * Set SMTP username - * - * @param string $username - * @return self - */ - public function setSmtpUsername(string $username): self - { - $this->smtp['username'] = $username; - return $this; - } - - /** - * Set SMTP password - * - * @param string $password - * @return self - */ - public function setSmtpPassword(string $password): self - { - $this->smtp['password'] = $password; - return $this; - } - - /** - * Set SMTP secure - * - * @param string $secure - * @return self - */ - public function setSmtpSecure(string $secure): self - { - $this->smtp['secure'] = $secure; - return $this; - } - - /** - * Set SMTP sender email - * - * @param string $senderEmail - * @return self - */ - public function setSmtpSenderEmail(string $senderEmail): self - { - $this->smtp['senderEmail'] = $senderEmail; - return $this; - } - - /** - * Set SMTP sender name - * - * @param string $senderName - * @return self - */ - public function setSmtpSenderName(string $senderName): self - { - $this->smtp['senderName'] = $senderName; - return $this; - } - - /** - * Set SMTP reply-to email - * - * @param string $email - * @return self - */ - public function setSmtpReplyToEmail(string $email): self - { - $this->smtp['replyToEmail'] = $email; - return $this; - } - - /** - * Set SMTP reply-to name - * - * @param string $name - * @return self - */ - public function setSmtpReplyToName(string $name): self - { - $this->smtp['replyToName'] = $name; - return $this; - } - - /** - * Get SMTP - * - * @return string - */ - public function getSmtpHost(): string - { - return $this->smtp['host'] ?? ''; - } - - /** - * Get SMTP port - * - * @return integer - */ - public function getSmtpPort(): int - { - return $this->smtp['port'] ?? 0; - } - - /** - * Get SMTP username - * - * @return string - */ - public function getSmtpUsername(): string - { - return $this->smtp['username'] ?? ''; - } - - /** - * Get SMTP password - * - * @return string - */ - public function getSmtpPassword(): string - { - return $this->smtp['password'] ?? ''; - } - - /** - * Get SMTP secure - * - * @return string - */ - public function getSmtpSecure(): string - { - return $this->smtp['secure'] ?? ''; - } - - /** - * Get SMTP sender email - * - * @return string - */ - public function getSmtpSenderEmail(): string - { - return $this->smtp['senderEmail'] ?? ''; - } - - /** - * Get SMTP sender name - * - * @return string - */ - public function getSmtpSenderName(): string - { - return $this->smtp['senderName'] ?? ''; - } - - /** - * Get SMTP reply-to email - * - * @return string - */ - public function getSmtpReplyToEmail(): string - { - return $this->smtp['replyToEmail'] ?? ''; - } - - /** - * Get SMTP reply-to name - * - * @return string - */ - public function getSmtpReplyToName(): string - { - return $this->smtp['replyToName'] ?? ''; - } - - /** - * Get Email Variables - * - * @return array - */ - public function getVariables(): array - { - return $this->variables; - } - - /** - * Set Email Variables - * - * @param array $variables - * @return self - */ - public function setVariables(array $variables): self - { - $this->variables = $variables; - return $this; - } - - /** - * Append variables to the email event. - * - * @param array $variables - * @return self - */ - public function appendVariables(array $variables): self - { - $this->variables = \array_merge($this->variables, $variables); - return $this; - } - - /** - * Set attachment - * @param string $content - * @param string $filename - * @param string $encoding - * @param string $type - * @return self - */ - public function setAttachment(string $content, string $filename, string $encoding = 'base64', string $type = 'plain/text') - { - $this->attachment = [ - 'content' => base64_encode($content), - 'filename' => $filename, - 'encoding' => $encoding, - 'type' => $type, - ]; - return $this; - } - - /** - * Get attachment - * - * @return array - */ - public function getAttachment(): array - { - return $this->attachment; - } - - /** - * Reset attachment - * - * @return self - */ - public function resetAttachment(): self - { - $this->attachment = []; - return $this; - } - - /** - * Set sender email - * - * @param string $email - * @return self - */ - public function setSenderEmail(string $email): self - { - $this->customMailOptions['senderEmail'] = $email; - return $this; - } - - /** - * Get sender email - * - * @return string - */ - public function getSenderEmail(): string - { - return $this->customMailOptions['senderEmail'] ?? ''; - } - - /** - * Set sender name - * - * @param string $name - * @return self - */ - public function setSenderName(string $name): self - { - $this->customMailOptions['senderName'] = $name; - return $this; - } - - /** - * Get sender name - * - * @return string - */ - public function getSenderName(): string - { - return $this->customMailOptions['senderName'] ?? ''; - } - - /** - * Set reply-to email - * - * @param string $email - * @return self - */ - public function setReplyToEmail(string $email): self - { - $this->customMailOptions['replyToEmail'] = $email; - return $this; - } - - /** - * Get reply-to email - * - * @return string - */ - public function getReplyToEmail(): string - { - return $this->customMailOptions['replyToEmail'] ?? ''; - } - - /** - * Set reply-to name - * - * @param string $name - * @return self - */ - public function setReplyToName(string $name): self - { - $this->customMailOptions['replyToName'] = $name; - return $this; - } - - /** - * Get reply-to name - * - * @return string - */ - public function getReplyToName(): string - { - return $this->customMailOptions['replyToName'] ?? ''; - } - - /** - * Reset - * - * @return self - */ - public function reset(): self - { - $this->project = null; - $this->recipient = ''; - $this->name = ''; - $this->subject = ''; - $this->body = ''; - $this->variables = []; - $this->bodyTemplate = ''; - $this->attachment = []; - $this->customMailOptions = []; - return $this; - } - - /** - * Prepare the payload for the event - * - * @return array - */ - protected function preparePayload(): array - { - $platform = $this->platform; - if (empty($platform)) { - $platform = Config::getParam('platform', []); - } - - return [ - 'project' => $this->project, - 'recipient' => $this->recipient, - 'name' => $this->name, - 'subject' => $this->subject, - 'bodyTemplate' => $this->bodyTemplate, - 'body' => $this->body, - 'preview' => $this->preview, - 'smtp' => $this->smtp, - 'variables' => $this->variables, - 'attachment' => $this->attachment, - 'customMailOptions' => $this->customMailOptions, - 'events' => Event::generateEvents($this->getEvent(), $this->getParams()), - 'platform' => $platform, - ]; - } -} diff --git a/src/Appwrite/Event/Message/Mail.php b/src/Appwrite/Event/Message/Mail.php new file mode 100644 index 0000000000..aeeea8a616 --- /dev/null +++ b/src/Appwrite/Event/Message/Mail.php @@ -0,0 +1,66 @@ +platform) ? $this->platform : Config::getParam('platform', []); + + return [ + 'project' => $this->project?->getArrayCopy(), + 'recipient' => $this->recipient, + 'name' => $this->name, + 'subject' => $this->subject, + 'bodyTemplate' => $this->bodyTemplate, + 'body' => $this->body, + 'preview' => $this->preview, + 'smtp' => $this->smtp, + 'variables' => $this->variables, + 'attachment' => $this->attachment, + 'customMailOptions' => $this->customMailOptions, + 'events' => $this->events, + 'platform' => $platform, + ]; + } + + public static function fromArray(array $data): static + { + return new self( + project: !empty($data['project']) ? new Document($data['project']) : null, + recipient: $data['recipient'] ?? '', + name: $data['name'] ?? '', + subject: $data['subject'] ?? '', + bodyTemplate: $data['bodyTemplate'] ?? '', + body: $data['body'] ?? '', + preview: $data['preview'] ?? '', + smtp: $data['smtp'] ?? [], + variables: $data['variables'] ?? [], + attachment: $data['attachment'] ?? [], + customMailOptions: $data['customMailOptions'] ?? [], + events: $data['events'] ?? [], + platform: $data['platform'] ?? [], + ); + } +} diff --git a/src/Appwrite/Event/Message/Messaging.php b/src/Appwrite/Event/Message/Messaging.php new file mode 100644 index 0000000000..7f0f918217 --- /dev/null +++ b/src/Appwrite/Event/Message/Messaging.php @@ -0,0 +1,45 @@ + $this->type, + 'project' => $this->project->getArrayCopy(), + 'user' => $this->user?->getArrayCopy(), + 'messageId' => $this->messageId, + 'message' => $this->message?->getArrayCopy(), + 'recipients' => $this->recipients, + 'providerType' => $this->providerType, + ]; + } + + public static function fromArray(array $data): static + { + return new self( + type: $data['type'] ?? '', + project: new Document($data['project'] ?? []), + user: !empty($data['user']) ? new Document($data['user']) : null, + messageId: $data['messageId'] ?? null, + message: !empty($data['message']) ? new Document($data['message']) : null, + recipients: $data['recipients'] ?? null, + providerType: $data['providerType'] ?? null, + ); + } +} diff --git a/src/Appwrite/Event/Messaging.php b/src/Appwrite/Event/Messaging.php deleted file mode 100644 index 9895d52ec2..0000000000 --- a/src/Appwrite/Event/Messaging.php +++ /dev/null @@ -1,182 +0,0 @@ -setQueue(System::getEnv('_APP_MESSAGING_QUEUE_NAME', Event::MESSAGING_QUEUE_NAME)) - ->setClass(System::getEnv('_APP_MESSAGING_CLASS_NAME', Event::MESSAGING_CLASS_NAME)); - } - - /** - * Sets type for the build event. - * - * @param string $type Can be `MESSAGE_SEND_TYPE_INTERNAL` or `MESSAGE_SEND_TYPE_EXTERNAL`. - * @return self - */ - public function setType(string $type): self - { - $this->type = $type; - - return $this; - } - - /** - * Returns set type for the function event. - * - * @return string - */ - public function getType(): string - { - return $this->type; - } - - /** - * Sets recipient for the messaging event. - * - * @param string[] $recipients - * @return self - */ - public function setRecipients(array $recipients): self - { - $this->recipients = $recipients; - - return $this; - } - - /** - * Returns set recipient for messaging event. - * - * @return string[] - */ - public function getRecipient(): array - { - return $this->recipients; - } - - /** - * Sets message document for the messaging event. - * - * @param Document $message - * @return self - */ - public function setMessage(Document $message): self - { - $this->message = $message; - - return $this; - } - - /** - * Returns message document for the messaging event. - * - * @return Document - */ - public function getMessage(): Document - { - return $this->message; - } - - /** - * Sets message ID for the messaging event. - * - * @param string $messageId - * @return self - */ - public function setMessageId(string $messageId): self - { - $this->messageId = $messageId; - - return $this; - } - - /** - * Returns set message ID for the messaging event. - * - * @return string - */ - public function getMessageId(): string - { - return $this->messageId; - } - - /** - * Sets provider type for the messaging event. - * - * @param string $providerType - * @return self - */ - public function setProviderType(string $providerType): self - { - $this->providerType = $providerType; - - return $this; - } - - /** - * Returns set provider type for the messaging event. - * - * @return string - */ - public function getProviderType(): string - { - return $this->providerType; - } - - /** - * Sets Scheduled delivery time for the messaging event. - * - * @param string $scheduledAt - * @return self - */ - public function setScheduledAt(string $scheduledAt): self - { - $this->scheduledAt = $scheduledAt; - - return $this; - } - - /** - * Returns set Delivery Time for the messaging event. - * - * @return string - */ - public function getScheduledAt(): string - { - return $this->scheduledAt; - } - - /** - * Prepare the payload for the event - * - * @return array - */ - protected function preparePayload(): array - { - return [ - 'type' => $this->type, - 'project' => $this->project, - 'user' => $this->user, - 'messageId' => $this->messageId, - 'message' => $this->message, - 'recipients' => $this->recipients, - 'providerType' => $this->providerType, - ]; - } -} diff --git a/src/Appwrite/Event/Publisher/Mail.php b/src/Appwrite/Event/Publisher/Mail.php new file mode 100644 index 0000000000..16d48be044 --- /dev/null +++ b/src/Appwrite/Event/Publisher/Mail.php @@ -0,0 +1,27 @@ +publish($queue ?? $this->queue, $message); + } + + public function getSize(bool $failed = false, ?Queue $queue = null): int + { + return $this->getQueueSize($queue ?? $this->queue, $failed); + } +} diff --git a/src/Appwrite/Event/Publisher/Messaging.php b/src/Appwrite/Event/Publisher/Messaging.php new file mode 100644 index 0000000000..69863566a1 --- /dev/null +++ b/src/Appwrite/Event/Publisher/Messaging.php @@ -0,0 +1,27 @@ +publish($queue ?? $this->queue, $message); + } + + public function getSize(bool $failed = false, ?Queue $queue = null): int + { + return $this->getQueueSize($queue ?? $this->queue, $failed); + } +} diff --git a/src/Appwrite/Platform/Modules/Account/Http/Account/MFA/Challenges/Create.php b/src/Appwrite/Platform/Modules/Account/Http/Account/MFA/Challenges/Create.php index 7bcc78e974..285875eb35 100644 --- a/src/Appwrite/Platform/Modules/Account/Http/Account/MFA/Challenges/Create.php +++ b/src/Appwrite/Platform/Modules/Account/Http/Account/MFA/Challenges/Create.php @@ -5,8 +5,10 @@ namespace Appwrite\Platform\Modules\Account\Http\Account\MFA\Challenges; use Appwrite\Auth\MFA\Type; use Appwrite\Detector\Detector; use Appwrite\Event\Event; -use Appwrite\Event\Mail; -use Appwrite\Event\Messaging; +use Appwrite\Event\Message\Mail as MailMessage; +use Appwrite\Event\Message\Messaging as MessagingMessage; +use Appwrite\Event\Publisher\Mail as MailPublisher; +use Appwrite\Event\Publisher\Messaging as MessagingPublisher; use Appwrite\Extend\Exception; use Appwrite\SDK\AuthType; use Appwrite\SDK\ContentType; @@ -101,8 +103,8 @@ class Create extends Action ->inject('platform') ->inject('request') ->inject('queueForEvents') - ->inject('queueForMessaging') - ->inject('queueForMails') + ->inject('publisherForMessaging') + ->inject('publisherForMails') ->inject('timelimit') ->inject('usage') ->inject('plan') @@ -121,8 +123,8 @@ class Create extends Action array $platform, Request $request, Event $queueForEvents, - Messaging $queueForMessaging, - Mail $queueForMails, + MessagingPublisher $publisherForMessaging, + MailPublisher $publisherForMails, callable $timelimit, Context $usage, array $plan, @@ -180,16 +182,18 @@ class Create extends Action $message = $message->render(); $phone = $user->getAttribute('phone'); - $queueForMessaging - ->setType(MESSAGE_SEND_TYPE_INTERNAL) - ->setMessage(new Document([ + $publisherForMessaging->enqueue(new MessagingMessage( + type: MESSAGE_SEND_TYPE_INTERNAL, + project: $project, + message: new Document([ '$id' => $challenge->getId(), 'data' => [ 'content' => $code, ], - ])) - ->setRecipients([$phone]) - ->setProviderType(MESSAGE_TYPE_SMS); + ]), + recipients: [$phone], + providerType: MESSAGE_TYPE_SMS, + )); $helper = PhoneNumberUtil::getInstance(); try { @@ -252,6 +256,7 @@ class Create extends Action $senderName = System::getEnv('_APP_SYSTEM_EMAIL_NAME', APP_NAME . ' Server'); $replyToEmail = ''; $replyToName = ''; + $smtpConfig = []; if ($smtpEnabled) { if (!empty($smtp['senderEmail'])) { @@ -269,13 +274,6 @@ class Create extends Action $replyToName = $smtp['replyToName']; } - $queueForMails - ->setSmtpHost($smtp['host'] ?? '') - ->setSmtpPort($smtp['port'] ?? '') - ->setSmtpUsername($smtp['username'] ?? '') - ->setSmtpPassword($smtp['password'] ?? '') - ->setSmtpSecure($smtp['secure'] ?? ''); - if (!empty($customTemplate)) { if (!empty($customTemplate['senderEmail'])) { $senderEmail = $customTemplate['senderEmail']; @@ -296,11 +294,17 @@ class Create extends Action $subject = $customTemplate['subject'] ?? $subject; } - $queueForMails - ->setSmtpReplyToEmail($replyToEmail) - ->setSmtpReplyToName($replyToName) - ->setSmtpSenderEmail($senderEmail) - ->setSmtpSenderName($senderName); + $smtpConfig = [ + 'host' => $smtp['host'] ?? '', + 'port' => $smtp['port'] ?? '', + 'username' => $smtp['username'] ?? '', + 'password' => $smtp['password'] ?? '', + 'secure' => $smtp['secure'] ?? '', + 'replyToEmail' => $replyToEmail, + 'replyToName' => $replyToName, + 'senderEmail' => $senderEmail, + 'senderName' => $senderName, + ]; } $emailVariables = [ @@ -327,20 +331,18 @@ class Create extends Action ]); } - $queueForMails - ->setSubject($subject) - ->setPreview($preview) - ->setBody($body) - ->setBodyTemplate($bodyTemplate) - ->appendVariables($emailVariables) - ->setRecipient($user->getAttribute('email')); - - // since this is console project, set email sender name! - if ($smtpBaseTemplate === APP_BRANDED_EMAIL_BASE_TEMPLATE) { - $queueForMails->setSenderName($platform['emailSenderName']); - } - - $queueForMails->trigger(); + $publisherForMails->enqueue(new MailMessage( + project: $project, + recipient: $user->getAttribute('email'), + subject: $subject, + bodyTemplate: $bodyTemplate, + body: $body, + preview: $preview, + smtp: $smtpConfig, + variables: $emailVariables, + customMailOptions: $smtpBaseTemplate === APP_BRANDED_EMAIL_BASE_TEMPLATE ? ['senderName' => $platform['emailSenderName']] : [], + platform: $platform, + )); break; } diff --git a/src/Appwrite/Platform/Modules/Health/Http/Health/Queue/Failed/Get.php b/src/Appwrite/Platform/Modules/Health/Http/Health/Queue/Failed/Get.php index 0d0a787b46..70d7713280 100644 --- a/src/Appwrite/Platform/Modules/Health/Http/Health/Queue/Failed/Get.php +++ b/src/Appwrite/Platform/Modules/Health/Http/Health/Queue/Failed/Get.php @@ -6,11 +6,11 @@ use Appwrite\Event\Database; use Appwrite\Event\Delete; use Appwrite\Event\Event; use Appwrite\Event\Func; -use Appwrite\Event\Mail; -use Appwrite\Event\Messaging; use Appwrite\Event\Publisher\Audit; use Appwrite\Event\Publisher\Build as BuildPublisher; use Appwrite\Event\Publisher\Certificate; +use Appwrite\Event\Publisher\Mail as MailPublisher; +use Appwrite\Event\Publisher\Messaging as MessagingPublisher; use Appwrite\Event\Publisher\Migration as MigrationPublisher; use Appwrite\Event\Publisher\Screenshot; use Appwrite\Event\Publisher\StatsResources as StatsResourcesPublisher; @@ -77,14 +77,14 @@ class Get extends Base ->inject('queueForDatabase') ->inject('queueForDeletes') ->inject('publisherForAudits') - ->inject('queueForMails') + ->inject('publisherForMails') ->inject('queueForFunctions') ->inject('publisherForStatsResources') ->inject('publisherForUsage') ->inject('queueForWebhooks') ->inject('publisherForCertificates') ->inject('publisherForBuilds') - ->inject('queueForMessaging') + ->inject('publisherForMessaging') ->inject('publisherForMigrations') ->inject('publisherForScreenshots') ->callback($this->action(...)); @@ -97,14 +97,14 @@ class Get extends Base Database $queueForDatabase, Delete $queueForDeletes, Audit $publisherForAudits, - Mail $queueForMails, + MailPublisher $publisherForMails, Func $queueForFunctions, StatsResourcesPublisher $publisherForStatsResources, UsagePublisher $publisherForUsage, Webhook $queueForWebhooks, Certificate $publisherForCertificates, BuildPublisher $publisherForBuilds, - Messaging $queueForMessaging, + MessagingPublisher $publisherForMessaging, MigrationPublisher $publisherForMigrations, Screenshot $publisherForScreenshots, ): void { @@ -114,7 +114,7 @@ class Get extends Base System::getEnv('_APP_DATABASE_QUEUE_NAME', Event::DATABASE_QUEUE_NAME) => $queueForDatabase, System::getEnv('_APP_DELETE_QUEUE_NAME', Event::DELETE_QUEUE_NAME) => $queueForDeletes, System::getEnv('_APP_AUDITS_QUEUE_NAME', Event::AUDITS_QUEUE_NAME) => $publisherForAudits, - System::getEnv('_APP_MAILS_QUEUE_NAME', Event::MAILS_QUEUE_NAME) => $queueForMails, + System::getEnv('_APP_MAILS_QUEUE_NAME', Event::MAILS_QUEUE_NAME) => $publisherForMails, System::getEnv('_APP_FUNCTIONS_QUEUE_NAME', Event::FUNCTIONS_QUEUE_NAME) => $queueForFunctions, System::getEnv('_APP_STATS_RESOURCES_QUEUE_NAME', Event::STATS_RESOURCES_QUEUE_NAME) => $publisherForStatsResources, System::getEnv('_APP_STATS_USAGE_QUEUE_NAME', Event::STATS_USAGE_QUEUE_NAME) => $publisherForUsage, @@ -122,7 +122,7 @@ class Get extends Base System::getEnv('_APP_CERTIFICATES_QUEUE_NAME', Event::CERTIFICATES_QUEUE_NAME) => $publisherForCertificates, System::getEnv('_APP_BUILDS_QUEUE_NAME', Event::BUILDS_QUEUE_NAME) => $publisherForBuilds, System::getEnv('_APP_SCREENSHOTS_QUEUE_NAME', Event::SCREENSHOTS_QUEUE_NAME) => $publisherForScreenshots, - System::getEnv('_APP_MESSAGING_QUEUE_NAME', Event::MESSAGING_QUEUE_NAME) => $queueForMessaging, + System::getEnv('_APP_MESSAGING_QUEUE_NAME', Event::MESSAGING_QUEUE_NAME) => $publisherForMessaging, System::getEnv('_APP_MIGRATIONS_QUEUE_NAME', Event::MIGRATIONS_QUEUE_NAME) => $publisherForMigrations, default => throw new Exception(Exception::GENERAL_SERVER_ERROR, 'Unknown queue name: ' . $name), }; diff --git a/src/Appwrite/Platform/Modules/Health/Http/Health/Queue/Mails/Get.php b/src/Appwrite/Platform/Modules/Health/Http/Health/Queue/Mails/Get.php index 3b9c06b5f9..2dd36e8111 100644 --- a/src/Appwrite/Platform/Modules/Health/Http/Health/Queue/Mails/Get.php +++ b/src/Appwrite/Platform/Modules/Health/Http/Health/Queue/Mails/Get.php @@ -2,7 +2,7 @@ namespace Appwrite\Platform\Modules\Health\Http\Health\Queue\Mails; -use Appwrite\Event\Mail; +use Appwrite\Event\Publisher\Mail as MailPublisher; use Appwrite\Platform\Modules\Health\Http\Health\Queue\Base; use Appwrite\SDK\AuthType; use Appwrite\SDK\ContentType; @@ -42,16 +42,16 @@ class Get extends Base contentType: ContentType::JSON )) ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true) - ->inject('queueForMails') + ->inject('publisherForMails') ->inject('response') ->callback($this->action(...)); } - public function action(int|string $threshold, Mail $queueForMails, Response $response): void + public function action(int|string $threshold, MailPublisher $publisherForMails, Response $response): void { $threshold = (int) $threshold; - $size = $queueForMails->getSize(); + $size = $publisherForMails->getSize(); $this->assertQueueThreshold($size, $threshold); diff --git a/src/Appwrite/Platform/Modules/Health/Http/Health/Queue/Messaging/Get.php b/src/Appwrite/Platform/Modules/Health/Http/Health/Queue/Messaging/Get.php index db2d7d7172..a2a829b1d5 100644 --- a/src/Appwrite/Platform/Modules/Health/Http/Health/Queue/Messaging/Get.php +++ b/src/Appwrite/Platform/Modules/Health/Http/Health/Queue/Messaging/Get.php @@ -2,7 +2,7 @@ namespace Appwrite\Platform\Modules\Health\Http\Health\Queue\Messaging; -use Appwrite\Event\Messaging; +use Appwrite\Event\Publisher\Messaging as MessagingPublisher; use Appwrite\Platform\Modules\Health\Http\Health\Queue\Base; use Appwrite\SDK\AuthType; use Appwrite\SDK\ContentType; @@ -42,16 +42,16 @@ class Get extends Base contentType: ContentType::JSON )) ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true) - ->inject('queueForMessaging') + ->inject('publisherForMessaging') ->inject('response') ->callback($this->action(...)); } - public function action(int|string $threshold, Messaging $queueForMessaging, Response $response): void + public function action(int|string $threshold, MessagingPublisher $publisherForMessaging, Response $response): void { $threshold = (int) $threshold; - $size = $queueForMessaging->getSize(); + $size = $publisherForMessaging->getSize(); $this->assertQueueThreshold($size, $threshold); diff --git a/src/Appwrite/Platform/Modules/Project/Http/Project/SMTP/Tests/Create.php b/src/Appwrite/Platform/Modules/Project/Http/Project/SMTP/Tests/Create.php index 7095c2d2d0..8c87a41475 100644 --- a/src/Appwrite/Platform/Modules/Project/Http/Project/SMTP/Tests/Create.php +++ b/src/Appwrite/Platform/Modules/Project/Http/Project/SMTP/Tests/Create.php @@ -2,7 +2,8 @@ namespace Appwrite\Platform\Modules\Project\Http\Project\SMTP\Tests; -use Appwrite\Event\Mail; +use Appwrite\Event\Message\Mail as MailMessage; +use Appwrite\Event\Publisher\Mail as MailPublisher; use Appwrite\Extend\Exception as Exception; use Appwrite\Platform\Action; use Appwrite\SDK\AuthType; @@ -67,7 +68,7 @@ class Create extends Action ->param('secure', '', new WhiteList(['tls', 'ssl'], true), 'Does SMTP server use secure connection', optional: true, deprecated: true) // Backwards compatibility ->inject('response') ->inject('project') - ->inject('queueForMails') + ->inject('publisherForMails') ->inject('plan') ->callback($this->action(...)); } @@ -87,7 +88,7 @@ class Create extends Action string $paramSecure, // Backwards compatibility Response $response, Document $project, - Mail $queueForMails, + MailPublisher $publisherForMails, array $plan ): void { // Backwards compatibility: use inline params if provided, otherwise fall back to project SMTP config. @@ -153,23 +154,24 @@ class Create extends Action ->setParam('{{privacyUrl}}', $plan['privacyUrl'] ?? APP_EMAIL_PRIVACY_URL); foreach ($emails as $email) { - $queueForMails - ->setSmtpHost($host) - ->setSmtpPort($port) - ->setSmtpUsername($username) - ->setSmtpPassword($password) - ->setSmtpSecure($secure) - ->setSmtpReplyToEmail($replyToEmail) - ->setSmtpReplyToName($replyToName) - ->setSmtpSenderEmail($senderEmail) - ->setSmtpSenderName($senderName) - ->setRecipient($email) - ->setName('') - ->setBodyTemplate(APP_CE_CONFIG_DIR . '/locale/templates/email-base-styled.tpl') - ->setBody($template->render()) - ->setVariables([]) - ->setSubject($subject) - ->trigger(); + $publisherForMails->enqueue(new MailMessage( + project: $project, + recipient: $email, + subject: $subject, + bodyTemplate: APP_CE_CONFIG_DIR . '/locale/templates/email-base-styled.tpl', + body: $template->render(), + smtp: [ + 'host' => $host, + 'port' => $port, + 'username' => $username, + 'password' => $password, + 'secure' => $secure, + 'replyToEmail' => $replyToEmail, + 'replyToName' => $replyToName, + 'senderEmail' => $senderEmail, + 'senderName' => $senderName, + ], + )); } $response->noContent(); diff --git a/src/Appwrite/Platform/Modules/Teams/Http/Memberships/Create.php b/src/Appwrite/Platform/Modules/Teams/Http/Memberships/Create.php index 51115b7861..d16a71780a 100644 --- a/src/Appwrite/Platform/Modules/Teams/Http/Memberships/Create.php +++ b/src/Appwrite/Platform/Modules/Teams/Http/Memberships/Create.php @@ -4,8 +4,10 @@ namespace Appwrite\Platform\Modules\Teams\Http\Memberships; use Appwrite\Auth\Validator\Phone; use Appwrite\Event\Event; -use Appwrite\Event\Mail; -use Appwrite\Event\Messaging; +use Appwrite\Event\Message\Mail as MailMessage; +use Appwrite\Event\Message\Messaging as MessagingMessage; +use Appwrite\Event\Publisher\Mail as MailPublisher; +use Appwrite\Event\Publisher\Messaging as MessagingPublisher; use Appwrite\Extend\Exception; use Appwrite\Platform\Action; use Appwrite\SDK\AuthType; @@ -87,8 +89,8 @@ class Create extends Action ->inject('dbForProject') ->inject('authorization') ->inject('locale') - ->inject('queueForMails') - ->inject('queueForMessaging') + ->inject('publisherForMails') + ->inject('publisherForMessaging') ->inject('queueForEvents') ->inject('timelimit') ->inject('usage') @@ -98,7 +100,7 @@ class Create extends Action ->callback($this->action(...)); } - public function action(string $teamId, string $email, string $userId, string $phone, array $roles, string $url, string $name, Response $response, Document $project, User $user, Database $dbForProject, Authorization $authorization, Locale $locale, Mail $queueForMails, Messaging $queueForMessaging, Event $queueForEvents, callable $timelimit, Context $usage, array $plan, Password $proofForPassword, Token $proofForToken) + public function action(string $teamId, string $email, string $userId, string $phone, array $roles, string $url, string $name, Response $response, Document $project, User $user, Database $dbForProject, Authorization $authorization, Locale $locale, MailPublisher $publisherForMails, MessagingPublisher $publisherForMessaging, Event $queueForEvents, callable $timelimit, Context $usage, array $plan, Password $proofForPassword, Token $proofForToken) { $isAppUser = $user->isApp($authorization->getRoles()); $isPrivilegedUser = $user->isPrivileged($authorization->getRoles()); @@ -345,6 +347,7 @@ class Create extends Action $senderName = System::getEnv('_APP_SYSTEM_EMAIL_NAME', APP_NAME . ' Server'); $replyToEmail = ''; $replyToName = ''; + $smtpConfig = []; if ($smtpEnabled) { if (! empty($smtp['senderEmail'])) { @@ -362,13 +365,6 @@ class Create extends Action $replyToName = $smtp['replyToName']; } - $queueForMails - ->setSmtpHost($smtp['host'] ?? '') - ->setSmtpPort($smtp['port'] ?? '') - ->setSmtpUsername($smtp['username'] ?? '') - ->setSmtpPassword($smtp['password'] ?? '') - ->setSmtpSecure($smtp['secure'] ?? ''); - if (! empty($customTemplate)) { if (! empty($customTemplate['senderEmail'])) { $senderEmail = $customTemplate['senderEmail']; @@ -389,11 +385,17 @@ class Create extends Action $subject = $customTemplate['subject'] ?? $subject; } - $queueForMails - ->setSmtpReplyToEmail($replyToEmail) - ->setSmtpReplyToName($replyToName) - ->setSmtpSenderEmail($senderEmail) - ->setSmtpSenderName($senderName); + $smtpConfig = [ + 'host' => $smtp['host'] ?? '', + 'port' => $smtp['port'] ?? '', + 'username' => $smtp['username'] ?? '', + 'password' => $smtp['password'] ?? '', + 'secure' => $smtp['secure'] ?? '', + 'replyToEmail' => $replyToEmail, + 'replyToName' => $replyToName, + 'senderEmail' => $senderEmail, + 'senderName' => $senderName, + ]; } $emailVariables = [ @@ -406,14 +408,16 @@ class Create extends Action 'project' => $projectName, ]; - $queueForMails - ->setSubject($subject) - ->setBody($body) - ->setPreview($preview) - ->setRecipient($invitee->getAttribute('email')) - ->setName($invitee->getAttribute('name', '')) - ->appendVariables($emailVariables) - ->trigger(); + $publisherForMails->enqueue(new MailMessage( + project: $project, + recipient: $invitee->getAttribute('email'), + name: $invitee->getAttribute('name', ''), + subject: $subject, + body: $body, + preview: $preview, + smtp: $smtpConfig, + variables: $emailVariables, + )); } elseif (! empty($phone)) { if (empty(System::getEnv('_APP_SMS_PROVIDER'))) { throw new Exception(Exception::GENERAL_PHONE_DISABLED, 'Phone provider not configured'); @@ -431,11 +435,13 @@ class Create extends Action ], ]); - $queueForMessaging - ->setType(MESSAGE_SEND_TYPE_INTERNAL) - ->setMessage($messageDoc) - ->setRecipients([$phone]) - ->setProviderType('SMS'); + $publisherForMessaging->enqueue(new MessagingMessage( + type: MESSAGE_SEND_TYPE_INTERNAL, + project: $project, + message: $messageDoc, + recipients: [$phone], + providerType: 'SMS', + )); $helper = PhoneNumberUtil::getInstance(); try { diff --git a/src/Appwrite/Platform/Tasks/ScheduleMessages.php b/src/Appwrite/Platform/Tasks/ScheduleMessages.php index 57f6dd8002..23068fcb9d 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleMessages.php +++ b/src/Appwrite/Platform/Tasks/ScheduleMessages.php @@ -2,8 +2,12 @@ namespace Appwrite\Platform\Tasks; -use Appwrite\Event\Messaging; +use Appwrite\Event\Event; +use Appwrite\Event\Message\Messaging as MessagingMessage; +use Appwrite\Event\Publisher\Messaging as MessagingPublisher; use Utopia\Database\Database; +use Utopia\Queue\Queue; +use Utopia\System\System; class ScheduleMessages extends ScheduleBase { @@ -40,15 +44,18 @@ class ScheduleMessages extends ScheduleBase } \go(function () use ($schedule, $scheduledAt, $dbForPlatform) { - $queueForMessaging = new Messaging($this->publisherMessaging); - $this->updateProjectAccess($schedule['project'], $dbForPlatform); - $queueForMessaging - ->setType(MESSAGE_SEND_TYPE_EXTERNAL) - ->setMessageId($schedule['resourceId']) - ->setProject($schedule['project']) - ->trigger(); + $publisherForMessaging = new MessagingPublisher( + $this->publisherMessaging, + new Queue(System::getEnv('_APP_MESSAGING_QUEUE_NAME', Event::MESSAGING_QUEUE_NAME)) + ); + + $publisherForMessaging->enqueue(new MessagingMessage( + type: MESSAGE_SEND_TYPE_EXTERNAL, + project: $schedule['project'], + messageId: $schedule['resourceId'], + )); $dbForPlatform->deleteDocument( 'schedules', diff --git a/src/Appwrite/Platform/Workers/Certificates.php b/src/Appwrite/Platform/Workers/Certificates.php index 4d04a3c92c..af3d145f85 100644 --- a/src/Appwrite/Platform/Workers/Certificates.php +++ b/src/Appwrite/Platform/Workers/Certificates.php @@ -5,8 +5,9 @@ namespace Appwrite\Platform\Workers; use Appwrite\Certificates\Adapter as CertificatesAdapter; use Appwrite\Event\Event; use Appwrite\Event\Func; -use Appwrite\Event\Mail; +use Appwrite\Event\Message\Mail as MailMessage; use Appwrite\Event\Publisher\Certificate; +use Appwrite\Event\Publisher\Mail as MailPublisher; use Appwrite\Event\Realtime; use Appwrite\Event\Webhook; use Appwrite\Extend\Exception as AppwriteException; @@ -50,7 +51,7 @@ class Certificates extends Action ->desc('Certificates worker') ->inject('message') ->inject('dbForPlatform') - ->inject('queueForMails') + ->inject('publisherForMails') ->inject('queueForEvents') ->inject('queueForWebhooks') ->inject('queueForFunctions') @@ -66,7 +67,7 @@ class Certificates extends Action /** * @param Message $message * @param Database $dbForPlatform - * @param Mail $queueForMails + * @param MailPublisher $publisherForMails * @param Event $queueForEvents * @param Webhook $queueForWebhooks * @param Func $queueForFunctions @@ -83,7 +84,7 @@ class Certificates extends Action public function action( Message $message, Database $dbForPlatform, - Mail $queueForMails, + MailPublisher $publisherForMails, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, @@ -116,7 +117,7 @@ class Certificates extends Action break; case \Appwrite\Event\Certificate::ACTION_GENERATION: - $this->handleCertificateGenerationAction($domain, $domainType, $dbForPlatform, $queueForMails, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime, $log, $certificates, $authorization, $skipRenewCheck, $plan, $validationDomain); + $this->handleCertificateGenerationAction($domain, $domainType, $dbForPlatform, $publisherForMails, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime, $log, $certificates, $authorization, $skipRenewCheck, $plan, $validationDomain); break; default: @@ -209,7 +210,7 @@ class Certificates extends Action * @param Domain $domain * @param ?string $domainType * @param Database $dbForPlatform - * @param Mail $queueForMails + * @param MailPublisher $publisherForMails * @param Event $queueForEvents * @param Webhook $queueForWebhooks * @param Func $queueForFunctions @@ -233,7 +234,7 @@ class Certificates extends Action Domain $domain, ?string $domainType, Database $dbForPlatform, - Mail $queueForMails, + MailPublisher $publisherForMails, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, @@ -358,7 +359,7 @@ class Certificates extends Action $rule->setAttribute('status', RULE_STATUS_CERTIFICATE_GENERATION_FAILED); // Send email to security email - $this->notifyError($domain->get(), $e->getMessage(), $attempts, $queueForMails, $plan); + $this->notifyError($domain->get(), $e->getMessage(), $attempts, $publisherForMails, $plan); throw $e; } finally { @@ -524,12 +525,12 @@ class Certificates extends Action * @param string $domain Domain that caused the error * @param string $errorMessage Verbose error message * @param int $attempt How many times it failed already - * @param Mail $queueForMails + * @param MailPublisher $publisherForMails * @param array $plan * @return void * @throws Exception */ - private function notifyError(string $domain, string $errorMessage, int $attempt, Mail $queueForMails, array $plan): void + private function notifyError(string $domain, string $errorMessage, int $attempt, MailPublisher $publisherForMails, array $plan): void { // Log error into console Console::warning('Cannot renew domain (' . $domain . ') on attempt no. ' . $attempt . ' certificate: ' . $errorMessage); @@ -560,14 +561,14 @@ class Certificates extends Action $subject = $locale->getText("emails.certificate.subject"); $preview = $locale->getText("emails.certificate.preview"); - $queueForMails - ->setSubject($subject) - ->setPreview($preview) - ->setBody($body) - ->setName('Appwrite Administrator') - ->setBodyTemplate(__DIR__ . '/../../../../app/config/locale/templates/email-base-styled.tpl') - ->setVariables($emailVariables) - ->setRecipient(System::getEnv('_APP_EMAIL_CERTIFICATES', System::getEnv('_APP_SYSTEM_SECURITY_EMAIL_ADDRESS'))) - ->trigger(); + $publisherForMails->enqueue(new MailMessage( + recipient: System::getEnv('_APP_EMAIL_CERTIFICATES', System::getEnv('_APP_SYSTEM_SECURITY_EMAIL_ADDRESS')), + name: 'Appwrite Administrator', + subject: $subject, + bodyTemplate: __DIR__ . '/../../../../app/config/locale/templates/email-base-styled.tpl', + body: $body, + preview: $preview, + variables: $emailVariables, + )); } } diff --git a/src/Appwrite/Platform/Workers/Migrations.php b/src/Appwrite/Platform/Workers/Migrations.php index b6c295b3bb..1d3f2f4622 100644 --- a/src/Appwrite/Platform/Workers/Migrations.php +++ b/src/Appwrite/Platform/Workers/Migrations.php @@ -3,9 +3,10 @@ namespace Appwrite\Platform\Workers; use Ahc\Jwt\JWT; -use Appwrite\Event\Mail; +use Appwrite\Event\Message\Mail as MailMessage; use Appwrite\Event\Message\Migration; use Appwrite\Event\Message\Usage as UsageMessage; +use Appwrite\Event\Publisher\Mail as MailPublisher; use Appwrite\Event\Publisher\Usage as UsagePublisher; use Appwrite\Event\Realtime; use Appwrite\Extend\Exception; @@ -102,7 +103,7 @@ class Migrations extends Action ->inject('queueForRealtime') ->inject('deviceForMigrations') ->inject('deviceForFiles') - ->inject('queueForMails') + ->inject('publisherForMails') ->inject('usage') ->inject('publisherForUsage') ->inject('plan') @@ -124,7 +125,7 @@ class Migrations extends Action Realtime $queueForRealtime, Device $deviceForMigrations, Device $deviceForFiles, - Mail $queueForMails, + MailPublisher $publisherForMails, Context $usage, UsagePublisher $publisherForUsage, array $plan, @@ -163,7 +164,7 @@ class Migrations extends Action $this->processMigration( $migration, $queueForRealtime, - $queueForMails, + $publisherForMails, $usage, $publisherForUsage, $platform, @@ -426,7 +427,7 @@ class Migrations extends Action protected function processMigration( Document $migration, Realtime $queueForRealtime, - Mail $queueForMails, + MailPublisher $publisherForMails, Context $usage, UsagePublisher $publisherForUsage, array $platform, @@ -630,7 +631,7 @@ class Migrations extends Action } $destination_type = $migration->getAttribute('destination'); if ($destination_type === DestinationCSV::getName() || $destination_type === DestinationJSON::getName()) { - $this->handleDataExportComplete($project, $migration, $queueForMails, $queueForRealtime, $platform, $authorization); + $this->handleDataExportComplete($project, $migration, $publisherForMails, $queueForRealtime, $platform, $authorization); } } finally { $source?->cleanup(); @@ -657,7 +658,7 @@ class Migrations extends Action * * @param Document $project * @param Document $migration - * @param Mail $queueForMails + * @param MailPublisher $publisherForMails * @param Realtime $queueForRealtime * @param array $platform * @param Authorization $authorization @@ -666,7 +667,7 @@ class Migrations extends Action protected function handleDataExportComplete( Document $project, Document $migration, - Mail $queueForMails, + MailPublisher $publisherForMails, Realtime $queueForRealtime, array $platform, Authorization $authorization, @@ -718,7 +719,7 @@ class Migrations extends Action project: $project, user: $user, options: $options, - queueForMails: $queueForMails, + publisherForMails: $publisherForMails, platform: $platform, exportType: $migration->getAttribute('destination') === DestinationJSON::getName() ? 'JSON' : 'CSV', sizeMB: $sizeMB @@ -781,7 +782,7 @@ class Migrations extends Action project: $project, user: $user, options: $options, - queueForMails: $queueForMails, + publisherForMails: $publisherForMails, platform: $platform, exportType: $migration->getAttribute('destination') === DestinationJSON::getName() ? 'JSON' : 'CSV', downloadUrl: $downloadUrl @@ -795,7 +796,7 @@ class Migrations extends Action * @param Document $project * @param Document $user The user who triggered the operation * @param array $options Migration options - * @param Mail $queueForMails + * @param MailPublisher $publisherForMails * @param array $platform * @param string $downloadUrl Download URL for successful exports * @param float $sizeMB File size in MB for failed exports @@ -807,7 +808,7 @@ class Migrations extends Action Document $project, Document $user, array $options, - Mail $queueForMails, + MailPublisher $publisherForMails, array $platform, string $exportType = 'CSV', string $downloadUrl = '', @@ -877,17 +878,18 @@ class Migrations extends Action 'type' => $exportType, ]; - $queueForMails - ->setProject($project) - ->setSubject($subject) - ->setPreview($preview) - ->setBody($emailBody) - ->setBodyTemplate(__DIR__ . '/../../../../app/config/locale/templates/email-base-styled.tpl') - ->setVariables($emailVariables) - ->setName($user->getAttribute('name', $user->getAttribute('email'))) - ->setRecipient($user->getAttribute('email')) - ->setSenderName($platform['emailSenderName']) - ->trigger(); + $publisherForMails->enqueue(new MailMessage( + project: $project, + recipient: $user->getAttribute('email'), + name: $user->getAttribute('name', $user->getAttribute('email')), + subject: $subject, + bodyTemplate: __DIR__ . '/../../../../app/config/locale/templates/email-base-styled.tpl', + body: $emailBody, + preview: $preview, + variables: $emailVariables, + customMailOptions: ['senderName' => $platform['emailSenderName']], + platform: $platform, + )); Console::info("CSV export {$emailType} notification email sent to " . $user->getAttribute('email')); } diff --git a/src/Appwrite/Platform/Workers/Webhooks.php b/src/Appwrite/Platform/Workers/Webhooks.php index a7f4595966..973e487de5 100644 --- a/src/Appwrite/Platform/Workers/Webhooks.php +++ b/src/Appwrite/Platform/Workers/Webhooks.php @@ -2,8 +2,9 @@ namespace Appwrite\Platform\Workers; -use Appwrite\Event\Mail; +use Appwrite\Event\Message\Mail as MailMessage; use Appwrite\Event\Message\Usage as UsageMessage; +use Appwrite\Event\Publisher\Mail as MailPublisher; use Appwrite\Event\Publisher\Usage as UsagePublisher; use Appwrite\Template\Template; use Appwrite\Usage\Context as UsageContext; @@ -36,7 +37,7 @@ class Webhooks extends Action ->inject('message') ->inject('project') ->inject('dbForPlatform') - ->inject('queueForMails') + ->inject('publisherForMails') ->inject('publisherForUsage') ->inject('log') ->inject('plan') @@ -47,14 +48,14 @@ class Webhooks extends Action * @param Message $message * @param Document $project * @param Database $dbForPlatform - * @param Mail $queueForMails + * @param MailPublisher $publisherForMails * @param UsagePublisher $publisherForUsage * @param Log $log * @param array $plan * @return void * @throws Exception */ - public function action(Message $message, Document $project, Database $dbForPlatform, Mail $queueForMails, UsagePublisher $publisherForUsage, Log $log, array $plan): void + public function action(Message $message, Document $project, Database $dbForPlatform, MailPublisher $publisherForMails, UsagePublisher $publisherForUsage, Log $log, array $plan): void { $this->errors = []; $payload = $message->getPayload(); @@ -73,7 +74,7 @@ class Webhooks extends Action foreach ($project->getAttribute('webhooks', []) as $webhook) { if (array_intersect($webhook->getAttribute('events', []), $events)) { - $this->execute($events, $webhookPayload, $webhook, $user, $project, $dbForPlatform, $queueForMails, $publisherForUsage, $plan); + $this->execute($events, $webhookPayload, $webhook, $user, $project, $dbForPlatform, $publisherForMails, $publisherForUsage, $plan); } } @@ -89,11 +90,11 @@ class Webhooks extends Action * @param Document $user * @param Document $project * @param Database $dbForPlatform - * @param Mail $queueForMails + * @param MailPublisher $publisherForMails * @param array $plan * @return void */ - private function execute(array $events, string $payload, Document $webhook, Document $user, Document $project, Database $dbForPlatform, Mail $queueForMails, UsagePublisher $publisherForUsage, array $plan): void + private function execute(array $events, string $payload, Document $webhook, Document $user, Document $project, Database $dbForPlatform, MailPublisher $publisherForMails, UsagePublisher $publisherForUsage, array $plan): void { if ($webhook->getAttribute('enabled') !== true) { return; @@ -171,7 +172,7 @@ class Webhooks extends Action if ($attempts >= \intval(System::getEnv('_APP_WEBHOOK_MAX_FAILED_ATTEMPTS', '10'))) { $webhook->setAttribute('enabled', false); $updatePayload['enabled'] = false; - $this->sendEmailAlert($attempts, $statusCode, $webhook, $project, $dbForPlatform, $queueForMails, $plan); + $this->sendEmailAlert($attempts, $statusCode, $webhook, $project, $dbForPlatform, $publisherForMails, $plan); } $dbForPlatform->updateDocument('webhooks', $webhook->getId(), new Document($updatePayload)); @@ -203,11 +204,11 @@ class Webhooks extends Action * @param Document $webhook * @param Document $project * @param Database $dbForPlatform - * @param Mail $queueForMails + * @param MailPublisher $publisherForMails * @param array $plan * @return void */ - public function sendEmailAlert(int $attempts, mixed $statusCode, Document $webhook, Document $project, Database $dbForPlatform, Mail $queueForMails, array $plan): void + public function sendEmailAlert(int $attempts, mixed $statusCode, Document $webhook, Document $project, Database $dbForPlatform, MailPublisher $publisherForMails, array $plan): void { $memberships = $dbForPlatform->find('memberships', [ Query::equal('teamInternalId', [$project->getAttribute('teamInternalId')]), @@ -251,18 +252,16 @@ class Webhooks extends Action ->setParam('{{message}}', $template->render()) ->setParam('{{year}}', date("Y")); - $queueForMails - ->setProject($project) - ->setSubject($subject) - ->setPreview($preview) - ->setBody($body->render()); - foreach ($users as $user) { - $queueForMails - ->setVariables(['user' => $user->getAttribute('name', '')]) - ->setName($user->getAttribute('name', '')) - ->setRecipient($user->getAttribute('email')) - ->trigger(); + $publisherForMails->enqueue(new MailMessage( + project: $project, + recipient: $user->getAttribute('email'), + name: $user->getAttribute('name', ''), + subject: $subject, + body: $body->render(), + preview: $preview, + variables: ['user' => $user->getAttribute('name', '')], + )); } } } From 18ec96f12417d70dc0273c439ea100e27c9cdcf9 Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Fri, 8 May 2026 14:41:58 +0530 Subject: [PATCH 2/3] Address Greptile feedback for queue publishers --- .../Modules/Teams/Http/Memberships/Create.php | 4 +++- src/Appwrite/Platform/Tasks/ScheduleMessages.php | 14 ++++++++------ 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/src/Appwrite/Platform/Modules/Teams/Http/Memberships/Create.php b/src/Appwrite/Platform/Modules/Teams/Http/Memberships/Create.php index d16a71780a..5500a56cbc 100644 --- a/src/Appwrite/Platform/Modules/Teams/Http/Memberships/Create.php +++ b/src/Appwrite/Platform/Modules/Teams/Http/Memberships/Create.php @@ -95,12 +95,13 @@ class Create extends Action ->inject('timelimit') ->inject('usage') ->inject('plan') + ->inject('platform') ->inject('proofForPassword') ->inject('proofForToken') ->callback($this->action(...)); } - public function action(string $teamId, string $email, string $userId, string $phone, array $roles, string $url, string $name, Response $response, Document $project, User $user, Database $dbForProject, Authorization $authorization, Locale $locale, MailPublisher $publisherForMails, MessagingPublisher $publisherForMessaging, Event $queueForEvents, callable $timelimit, Context $usage, array $plan, Password $proofForPassword, Token $proofForToken) + public function action(string $teamId, string $email, string $userId, string $phone, array $roles, string $url, string $name, Response $response, Document $project, User $user, Database $dbForProject, Authorization $authorization, Locale $locale, MailPublisher $publisherForMails, MessagingPublisher $publisherForMessaging, Event $queueForEvents, callable $timelimit, Context $usage, array $plan, array $platform, Password $proofForPassword, Token $proofForToken) { $isAppUser = $user->isApp($authorization->getRoles()); $isPrivilegedUser = $user->isPrivileged($authorization->getRoles()); @@ -417,6 +418,7 @@ class Create extends Action preview: $preview, smtp: $smtpConfig, variables: $emailVariables, + platform: $platform, )); } elseif (! empty($phone)) { if (empty(System::getEnv('_APP_SMS_PROVIDER'))) { diff --git a/src/Appwrite/Platform/Tasks/ScheduleMessages.php b/src/Appwrite/Platform/Tasks/ScheduleMessages.php index 23068fcb9d..634fb26dc2 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleMessages.php +++ b/src/Appwrite/Platform/Tasks/ScheduleMessages.php @@ -14,6 +14,8 @@ class ScheduleMessages extends ScheduleBase public const UPDATE_TIMER = 3; // seconds public const ENQUEUE_TIMER = 4; // seconds + private ?MessagingPublisher $publisherForMessaging = null; + public static function getName(): string { return 'schedule-messages'; @@ -31,6 +33,11 @@ class ScheduleMessages extends ScheduleBase protected function enqueueResources(Database $dbForPlatform, callable $getProjectDB): void { + $publisherForMessaging = $this->publisherForMessaging ??= new MessagingPublisher( + $this->publisherMessaging, + new Queue(System::getEnv('_APP_MESSAGING_QUEUE_NAME', Event::MESSAGING_QUEUE_NAME)) + ); + foreach ($this->schedules as $schedule) { if (!$schedule['active']) { continue; @@ -43,14 +50,9 @@ class ScheduleMessages extends ScheduleBase continue; } - \go(function () use ($schedule, $scheduledAt, $dbForPlatform) { + \go(function () use ($schedule, $scheduledAt, $dbForPlatform, $publisherForMessaging) { $this->updateProjectAccess($schedule['project'], $dbForPlatform); - $publisherForMessaging = new MessagingPublisher( - $this->publisherMessaging, - new Queue(System::getEnv('_APP_MESSAGING_QUEUE_NAME', Event::MESSAGING_QUEUE_NAME)) - ); - $publisherForMessaging->enqueue(new MessagingMessage( type: MESSAGE_SEND_TYPE_EXTERNAL, project: $schedule['project'], From cf1bb1a1cc4e93fe869d54507466700574d73bb9 Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Mon, 11 May 2026 17:59:43 +0530 Subject: [PATCH 3/3] build