diff --git a/.env b/.env
index cb0ae9a424..aa90470729 100644
--- a/.env
+++ b/.env
@@ -103,8 +103,8 @@ _APP_MESSAGE_SMS_PROVIDER_MSG91_SENDER_ID=
 _APP_MESSAGE_SMS_PROVIDER_MSG91_AUTH_KEY=
 _APP_MESSAGE_SMS_PROVIDER_MSG91_FROM=
 _APP_MESSAGE_SMS_PROVIDER_MSG91_TO=
-_APP_MESSAGE_SMS_PROVIDER_MAILGUN_API_KEY=
-_APP_MESSAGE_SMS_PROVIDER_MAILGUN_DOMAIN=
-_APP_MESSAGE_SMS_PROVIDER_MAILGUN_FROM=
-_APP_MESSAGE_SMS_PROVIDER_MAILGUN_RECEIVER_EMAIL=
-_APP_MESSAGE_SMS_PROVIDER_MAILGUN_IS_EU_REGION=
+_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_API_KEY=
+_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_DOMAIN=
+_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_FROM=
+_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_RECEIVER_EMAIL=
+_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_IS_EU_REGION=
diff --git a/app/config/errors.php b/app/config/errors.php
index 8001517d78..5514e8e68e 100644
--- a/app/config/errors.php
+++ b/app/config/errors.php
@@ -756,10 +756,39 @@ return [
         'code' => 400,
     ],
 
+    /** Topic Errors */
+    Exception::TOPIC_NOT_FOUND => [
+        'name' => Exception::TOPIC_NOT_FOUND,
+        'description' => 'Topic with the request ID could not be found.',
+        'code' => 404,
+    ],
+    Exception::TOPIC_ALREADY_EXISTS => [
+        'name' => Exception::TOPIC_ALREADY_EXISTS,
+        'description' => 'Topic with the request ID already exists.',
+        'code' => 409,
+    ],
+
+    /** Subscriber Errors */
+    Exception::SUBSCRIBER_NOT_FOUND => [
+        'name' => Exception::SUBSCRIBER_NOT_FOUND,
+        'description' => 'Subscriber with the request ID could not be found.',
+        'code' => 404,
+    ],
+    Exception::SUBSCRIBER_ALREADY_EXISTS => [
+        'name' => Exception::SUBSCRIBER_ALREADY_EXISTS,
+        'description' => 'Subscriber with the request ID already exists.',
+        'code' => 409,
+    ],
+
     /** Message Errors */
     Exception::MESSAGE_NOT_FOUND => [
         'name' => Exception::MESSAGE_NOT_FOUND,
         'description' => 'Message with the requested ID could not be found.',
         'code' => 404,
+    ],
+    Exception::MESSAGE_ALREADY_SENT => [
+        'name' => Exception::MESSAGE_ALREADY_SENT,
+        'description' => 'Message with the requested ID has already been sent.',
+        'code' => 400,
     ]
 ];
diff --git a/app/controllers/api/databases.php b/app/controllers/api/databases.php
index 0c60c01a1e..9eb243521b 100644
--- a/app/controllers/api/databases.php
+++ b/app/controllers/api/databases.php
@@ -2489,8 +2489,11 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/indexes')
             'orders' => $orders,
         ]);
 
-        $validator = new IndexValidator($dbForProject->getAdapter()->getMaxIndexLength());
-        if (!$validator->isValid($collection->setAttribute('indexes', $index, Document::SET_TYPE_APPEND))) {
+        $validator = new IndexValidator(
+            $collection->getAttribute('attributes'),
+            $dbForProject->getAdapter()->getMaxIndexLength()
+        );
+        if (!$validator->isValid($index)) {
             throw new Exception(Exception::INDEX_INVALID, $validator->getDescription());
         }
 
diff --git a/app/controllers/api/messaging.php b/app/controllers/api/messaging.php
index 533086a058..72f0b9fdb4 100644
--- a/app/controllers/api/messaging.php
+++ b/app/controllers/api/messaging.php
@@ -1,11 +1,15 @@
 <?php
 
+use Appwrite\Event\Delete;
 use Appwrite\Event\Messaging;
 use Appwrite\Extend\Exception;
 use Appwrite\Permission;
 use Appwrite\Role;
 use Appwrite\Utopia\Database\Validator\CustomId;
+use Appwrite\Utopia\Database\Validator\Queries\Messages;
 use Appwrite\Utopia\Database\Validator\Queries\Providers;
+use Appwrite\Utopia\Database\Validator\Queries\Subscribers;
+use Appwrite\Utopia\Database\Validator\Queries\Topics;
 use Appwrite\Utopia\Response;
 use Utopia\App;
 use Utopia\Database\Database;
@@ -18,10 +22,8 @@ use Utopia\Database\Validator\Datetime as DatetimeValidator;
 use Utopia\Database\Validator\UID;
 use Utopia\Validator\ArrayList;
 use Utopia\Validator\Boolean;
-use Utopia\Validator\JSON;
 use Utopia\Validator\Text;
 use Utopia\Validator\WhiteList;
-use Utopia\Database\DateTime;
 
 App::post('/v1/messaging/providers/mailgun')
     ->desc('Create Mailgun Provider')
@@ -582,11 +584,9 @@ App::get('/v1/messaging/providers')
 
         if ($cursor) {
             $providerId = $cursor->getValue();
-            $cursorDocument = Authorization::skip(fn () => $dbForProject->findOne('providers', [
-                Query::equal('$id', [$providerId]),
-            ]));
+            $cursorDocument = Authorization::skip(fn () => $dbForProject->getDocument('providers', $providerId));
 
-            if ($cursorDocument === false || $cursorDocument->isEmpty()) {
+            if ($cursorDocument->isEmpty()) {
                 throw new Exception(Exception::GENERAL_CURSOR_NOT_FOUND, "Provider '{$providerId}' for the 'cursor' value not found.");
             }
 
@@ -1192,19 +1192,386 @@ App::delete('/v1/messaging/providers/:providerId')
 
         $dbForProject->deleteDocument('providers', $provider->getId());
 
-        $response->noContent();
+        $response
+            ->setStatusCode(Response::STATUS_CODE_NOCONTENT)
+            ->noContent();
+    });
+
+App::post('/v1/messaging/topics')
+    ->desc('Create a topic.')
+    ->groups(['api', 'messaging'])
+    ->label('audits.event', 'topics.create')
+    ->label('audits.resource', 'topics/{response.$id}')
+    ->label('scope', 'topics.write')
+    ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY])
+    ->label('sdk.namespace', 'messaging')
+    ->label('sdk.method', 'createTopic')
+    ->label('sdk.description', '/docs/references/messaging/create-topic.md')
+    ->label('sdk.response.code', Response::STATUS_CODE_CREATED)
+    ->label('sdk.response.type', Response::CONTENT_TYPE_JSON)
+    ->label('sdk.response.model', Response::MODEL_TOPIC)
+    ->param('topicId', '', new CustomId(), 'Topic ID. Choose a custom Topic ID or a new Topic ID.')
+    ->param('providerId', '', new UID(), 'Provider ID.')
+    ->param('name', '', new Text(128), 'Topic Name.')
+    ->param('description', '', new Text(2048), 'Topic Description.', true)
+    ->inject('dbForProject')
+    ->inject('response')
+    ->action(function (string $topicId, string $providerId, string $name, string $description, Database $dbForProject, Response $response) {
+        $topicId = $topicId == 'unique()' ? ID::unique() : $topicId;
+        $provider = $dbForProject->getDocument('providers', $providerId);
+
+        if ($provider->isEmpty()) {
+            throw new Exception(Exception::PROVIDER_NOT_FOUND);
+        }
+
+        $topic = new Document([
+            '$id' => $topicId,
+            'providerId' => $providerId,
+            'providerInternalId' => $provider->getInternalId(),
+            'name' => $name,
+        ]);
+
+        if ($description) {
+            $topic->setAttribute('description', $description);
+        }
+
+        try {
+            $topic = $dbForProject->createDocument('topics', $topic);
+        } catch (DuplicateException) {
+            throw new Exception(Exception::TOPIC_ALREADY_EXISTS);
+        }
+
+        $response
+            ->setStatusCode(Response::STATUS_CODE_CREATED)
+            ->dynamic($topic, Response::MODEL_TOPIC);
+    });
+
+App::get('/v1/messaging/topics')
+    ->desc('List topics.')
+    ->groups(['api', 'messaging'])
+    ->label('scope', 'topics.read')
+    ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY])
+    ->label('sdk.namespace', 'messaging')
+    ->label('sdk.method', 'listTopics')
+    ->label('sdk.description', '/docs/references/messaging/list-topics.md')
+    ->label('sdk.response.code', Response::STATUS_CODE_OK)
+    ->label('sdk.response.type', Response::CONTENT_TYPE_JSON)
+    ->label('sdk.response.model', Response::MODEL_TOPIC_LIST)
+    ->param('queries', [], new Topics(), 'Array of query strings generated using the Query class provided by the SDK. [Learn more about queries](https://appwrite.io/docs/queries). Maximum of ' . APP_LIMIT_ARRAY_PARAMS_SIZE . ' queries are allowed, each ' . APP_LIMIT_ARRAY_ELEMENT_SIZE . ' characters long. You may filter on the following attributes: ' . implode(', ', Topics::ALLOWED_ATTRIBUTES), true)
+    ->inject('dbForProject')
+    ->inject('response')
+    ->action(function (array $queries, Database $dbForProject, Response $response) {
+        $queries = Query::parseQueries($queries);
+
+        // Get cursor document if there was a cursor query
+        $cursor = Query::getByType($queries, [Query::TYPE_CURSORAFTER, Query::TYPE_CURSORBEFORE]);
+        $cursor = reset($cursor);
+
+        if ($cursor) {
+            $topicId = $cursor->getValue();
+            $cursorDocument = Authorization::skip(fn () => $dbForProject->getDocument('topics', $topicId));
+
+            if ($cursorDocument->isEmpty()) {
+                throw new Exception(Exception::GENERAL_CURSOR_NOT_FOUND, "Topic '{$topicId}' for the 'cursor' value not found.");
+            }
+
+            $cursor->setValue($cursorDocument[0]);
+        }
+
+        $filterQueries = Query::groupByType($queries)['filters'];
+        $response->dynamic(new Document([
+            'total' => $dbForProject->count('topics', $filterQueries, APP_LIMIT_COUNT),
+            'topics' => $dbForProject->find('topics', $queries),
+        ]), Response::MODEL_TOPIC_LIST);
+    });
+
+App::get('/v1/messaging/topics/:topicId')
+    ->desc('Get a topic.')
+    ->groups(['api', 'messaging'])
+    ->label('scope', 'topics.read')
+    ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY])
+    ->label('sdk.namespace', 'messaging')
+    ->label('sdk.method', 'getTopic')
+    ->label('sdk.description', '/docs/references/messaging/get-topic.md')
+    ->label('sdk.response.code', Response::STATUS_CODE_OK)
+    ->label('sdk.response.type', Response::CONTENT_TYPE_JSON)
+    ->label('sdk.response.model', Response::MODEL_TOPIC)
+    ->param('topicId', '', new UID(), 'Topic ID.')
+    ->inject('dbForProject')
+    ->inject('response')
+    ->action(function (string $topicId, Database $dbForProject, Response $response) {
+        $topic = $dbForProject->getDocument('topics', $topicId);
+
+        if ($topic->isEmpty()) {
+            throw new Exception(Exception::TOPIC_NOT_FOUND);
+        }
+
+        $topic = $dbForProject->getDocument('topics', $topicId);
+
+        $response
+            ->dynamic($topic, Response::MODEL_TOPIC);
+    });
+
+App::patch('/v1/messaging/topics/:topicId')
+    ->desc('Update a topic.')
+    ->groups(['api', 'messaging'])
+    ->label('audits.event', 'topics.update')
+    ->label('audits.resource', 'topics/{response.$id}')
+    ->label('scope', 'topics.write')
+    ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY])
+    ->label('sdk.namespace', 'messaging')
+    ->label('sdk.method', 'updateTopic')
+    ->label('sdk.description', '/docs/references/messaging/update-topic.md')
+    ->label('sdk.response.code', Response::STATUS_CODE_OK)
+    ->label('sdk.response.type', Response::CONTENT_TYPE_JSON)
+    ->label('sdk.response.model', Response::MODEL_TOPIC)
+    ->param('topicId', '', new UID(), 'Topic ID.')
+    ->param('name', '', new Text(128), 'Topic Name.', true)
+    ->param('description', '', new Text(2048), 'Topic Description.', true)
+    ->inject('dbForProject')
+    ->inject('response')
+    ->action(function (string $topicId, string $name, string $description, Database $dbForProject, Response $response) {
+        $topic = $dbForProject->getDocument('topics', $topicId);
+
+        if ($topic->isEmpty()) {
+            throw new Exception(Exception::TOPIC_NOT_FOUND);
+        }
+
+        if (!empty($name)) {
+            $topic->setAttribute('name', $name);
+        }
+
+        if (!empty($description)) {
+            $topic->setAttribute('description', $description);
+        }
+
+        $topic = $dbForProject->updateDocument('topics', $topicId, $topic);
+
+        $response
+            ->dynamic($topic, Response::MODEL_TOPIC);
+    });
+
+App::delete('/v1/messaging/topics/:topicId')
+    ->desc('Delete a topic.')
+    ->groups(['api', 'messaging'])
+    ->label('audits.event', 'topics.delete')
+    ->label('audits.resource', 'topics/{request.topicId}')
+    ->label('scope', 'topics.write')
+    ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY])
+    ->label('sdk.namespace', 'messaging')
+    ->label('sdk.method', 'deleteTopic')
+    ->label('sdk.description', '/docs/references/messaging/delete-topic.md')
+    ->label('sdk.response.code', Response::STATUS_CODE_NOCONTENT)
+    ->label('sdk.response.type', Response::CONTENT_TYPE_JSON)
+    ->label('sdk.response.model', Response::MODEL_NONE)
+    ->param('topicId', '', new UID(), 'Topic ID.')
+    ->inject('dbForProject')
+    ->inject('deletes')
+    ->inject('response')
+    ->action(function (string $topicId, Database $dbForProject, Delete $deletes, Response $response) {
+        $topic = $dbForProject->getDocument('topics', $topicId);
+
+        if ($topic->isEmpty()) {
+            throw new Exception(Exception::TOPIC_NOT_FOUND);
+        }
+
+        $dbForProject->deleteDocument('topics', $topicId);
+
+        $deletes
+            ->setType(DELETE_TYPE_SUBSCRIBERS)
+            ->setDocument($topic);
+
+        $response
+            ->setStatusCode(Response::STATUS_CODE_NOCONTENT)
+            ->noContent();
+    });
+
+App::post('/v1/messaging/topics/:topicId/subscribers')
+    ->desc('Adds a Subscriber to a Topic.')
+    ->groups(['api', 'messaging'])
+    ->label('audits.event', 'subscribers.create')
+    ->label('audits.resource', 'subscribers/{response.$id}')
+    ->label('scope', 'subscribers.write')
+    ->label('sdk.auth', [APP_AUTH_TYPE_JWT, APP_AUTH_TYPE_SESSION, APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY])
+    ->label('sdk.namespace', 'messaging')
+    ->label('sdk.method', 'createSubscriber')
+    ->label('sdk.description', '/docs/references/messaging/create-subscriber.md')
+    ->label('sdk.response.code', Response::STATUS_CODE_CREATED)
+    ->label('sdk.response.type', Response::CONTENT_TYPE_JSON)
+    ->label('sdk.response.model', Response::MODEL_SUBSCRIBER)
+    ->param('subscriberId', '', new CustomId(), 'Subscriber ID. Choose a custom Topic ID or a new Topic ID.')
+    ->param('topicId', '', new UID(), 'Topic ID.')
+    ->param('targetId', '', new UID(), 'Target ID.')
+    ->inject('dbForProject')
+    ->inject('response')
+    ->action(function (string $subscriberId, string $topicId, string $targetId, Database $dbForProject, Response $response) {
+        $subscriberId = $subscriberId == 'unique()' ? ID::unique() : $subscriberId;
+
+        $topic = Authorization::skip(fn () => $dbForProject->getDocument('topics', $topicId));
+
+        if ($topic->isEmpty()) {
+            throw new Exception(Exception::TOPIC_NOT_FOUND);
+        }
+
+        $target = Authorization::skip(fn () => $dbForProject->getDocument('targets', $targetId));
+
+        if ($target->isEmpty()) {
+            throw new Exception(Exception::USER_TARGET_NOT_FOUND);
+        }
+
+        $subscriber = new Document([
+            '$id' => $subscriberId,
+            '$permissions' => [
+                Permission::read(Role::user($target->getAttribute('userId'))),
+                Permission::delete(Role::user($target->getAttribute('userId'))),
+            ],
+            'topicId' => $topicId,
+            'topicInternalId' => $topic->getInternalId(),
+            'targetId' => $targetId,
+            'targetInternalId' => $target->getInternalId(),
+        ]);
+
+        try {
+            $subscriber = $dbForProject->createDocument('subscribers', $subscriber);
+            $dbForProject->deleteCachedDocument('topics', $topicId);
+        } catch (DuplicateException) {
+            throw new Exception(Exception::SUBSCRIBER_ALREADY_EXISTS);
+        }
+
+        $response
+            ->setStatusCode(Response::STATUS_CODE_CREATED)
+            ->dynamic($subscriber, Response::MODEL_SUBSCRIBER);
+    });
+
+App::get('/v1/messaging/topics/:topicId/subscribers')
+    ->desc('List topic\'s subscribers.')
+    ->groups(['api', 'messaging'])
+    ->label('scope', 'subscribers.read')
+    ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY])
+    ->label('sdk.namespace', 'messaging')
+    ->label('sdk.method', 'listSubscribers')
+    ->label('sdk.description', '/docs/references/messaging/list-subscribers.md')
+    ->label('sdk.response.code', Response::STATUS_CODE_OK)
+    ->label('sdk.response.type', Response::CONTENT_TYPE_JSON)
+    ->label('sdk.response.model', Response::MODEL_SUBSCRIBER_LIST)
+    ->param('topicId', '', new UID(), 'Topic ID.')
+    ->param('queries', [], new Subscribers(), 'Array of query strings generated using the Query class provided by the SDK. [Learn more about queries](https://appwrite.io/docs/queries). Maximum of ' . APP_LIMIT_ARRAY_PARAMS_SIZE . ' queries are allowed, each ' . APP_LIMIT_ARRAY_ELEMENT_SIZE . ' characters long. You may filter on the following attributes: ' . implode(', ', Providers::ALLOWED_ATTRIBUTES), true)
+    ->inject('dbForProject')
+    ->inject('response')
+    ->action(function (string $topicId, array $queries, Database $dbForProject, Response $response) {
+        $queries = Query::parseQueries($queries);
+
+        $topic = Authorization::skip(fn () => $dbForProject->getDocument('topics', $topicId));
+
+        if ($topic->isEmpty()) {
+            throw new Exception(Exception::TOPIC_NOT_FOUND);
+        }
+
+        \array_push($queries, Query::equal('topicInternalId', [$topic->getInternalId()]));
+
+        // Get cursor document if there was a cursor query
+        $cursor = Query::getByType($queries, [Query::TYPE_CURSORAFTER, Query::TYPE_CURSORBEFORE]);
+        $cursor = reset($cursor);
+
+        if ($cursor) {
+            $subscriberId = $cursor->getValue();
+            $cursorDocument = Authorization::skip(fn () => $dbForProject->getDocument('subscribers', $subscriberId));
+
+            if ($cursorDocument->isEmpty()) {
+                throw new Exception(Exception::GENERAL_CURSOR_NOT_FOUND, "Subscriber '{$subscriberId}' for the 'cursor' value not found.");
+            }
+
+            $cursor->setValue($cursorDocument);
+        }
+
+        $filterQueries = Query::groupByType($queries)['filters'];
+
+        $response
+            ->dynamic(new Document([
+                'subscribers' => $dbForProject->find('subscribers', $queries),
+                'total' => $dbForProject->count('subscribers', $filterQueries, APP_LIMIT_COUNT),
+            ]), Response::MODEL_SUBSCRIBER_LIST);
+    });
+
+App::get('/v1/messaging/topics/:topicId/subscriber/:subscriberId')
+    ->desc('Get a topic\'s subscriber.')
+    ->groups(['api', 'messaging'])
+    ->label('scope', 'subscribers.read')
+    ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY])
+    ->label('sdk.namespace', 'messaging')
+    ->label('sdk.method', 'getSubscriber')
+    ->label('sdk.description', '/docs/references/messaging/get-subscriber.md')
+    ->label('sdk.response.code', Response::STATUS_CODE_OK)
+    ->label('sdk.response.type', Response::CONTENT_TYPE_JSON)
+    ->label('sdk.response.model', Response::MODEL_SUBSCRIBER)
+    ->param('topicId', '', new UID(), 'Topic ID.')
+    ->param('subscriberId', '', new UID(), 'Subscriber ID.')
+    ->inject('dbForProject')
+    ->inject('response')
+    ->action(function (string $topicId, string $subscriberId, Database $dbForProject, Response $response) {
+        $topic = Authorization::skip(fn () => $dbForProject->getDocument('topics', $topicId));
+
+        if ($topic->isEmpty()) {
+            throw new Exception(Exception::TOPIC_NOT_FOUND);
+        }
+
+        $subscriber = $dbForProject->getDocument('subscribers', $subscriberId);
+
+        if ($subscriber->isEmpty() || $subscriber->getAttribute('topicId') !== $topicId) {
+            throw new Exception(Exception::SUBSCRIBER_NOT_FOUND);
+        }
+
+        $response
+            ->dynamic($subscriber, Response::MODEL_SUBSCRIBER);
+    });
+
+App::delete('/v1/messaging/topics/:topicId/subscriber/:subscriberId')
+    ->desc('Delete a Subscriber from a Topic.')
+    ->groups(['api', 'messaging'])
+    ->label('audits.event', 'subscribers.delete')
+    ->label('audits.resource', 'subscribers/{request.subscriberId}')
+    ->label('scope', 'subscribers.write')
+    ->label('sdk.auth', [APP_AUTH_TYPE_JWT, APP_AUTH_TYPE_SESSION, APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY])
+    ->label('sdk.namespace', 'messaging')
+    ->label('sdk.method', 'deleteSubscriber')
+    ->label('sdk.description', '/docs/references/messaging/delete-subscriber.md')
+    ->label('sdk.response.code', Response::STATUS_CODE_NOCONTENT)
+    ->label('sdk.response.type', Response::CONTENT_TYPE_JSON)
+    ->label('sdk.response.model', Response::MODEL_NONE)
+    ->param('topicId', '', new UID(), 'Topic ID.')
+    ->param('subscriberId', '', new UID(), 'Subscriber ID.')
+    ->inject('dbForProject')
+    ->inject('response')
+    ->action(function (string $topicId, string $subscriberId, Database $dbForProject, Response $response) {
+        $topic = Authorization::skip(fn () => $dbForProject->getDocument('topics', $topicId));
+
+        if ($topic->isEmpty()) {
+            throw new Exception(Exception::TOPIC_NOT_FOUND);
+        }
+
+        $subscriber = $dbForProject->getDocument('subscribers', $subscriberId);
+
+        if ($subscriber->isEmpty() || $subscriber->getAttribute('topicId') !== $topicId) {
+            throw new Exception(Exception::SUBSCRIBER_NOT_FOUND);
+        }
+        $subscriber = $dbForProject->deleteDocument('subscribers', $subscriberId);
+        $dbForProject->deleteCachedDocument('topics', $topicId);
+
+        $response
+            ->setStatusCode(Response::STATUS_CODE_NOCONTENT)
+            ->noContent();
     });
 
 App::post('/v1/messaging/messages/email')
-    ->desc('Send an email.')
+    ->desc('Create an email.')
     ->groups(['api', 'messaging'])
     ->label('audits.event', 'messages.create')
     ->label('audits.resource', 'messages/{response.$id}')
     ->label('scope', 'messages.write')
     ->label('sdk.auth', [APP_AUTH_TYPE_ADMIN, APP_AUTH_TYPE_KEY])
     ->label('sdk.namespace', 'messaging')
-    ->label('sdk.method', 'sendEmail')
-    ->label('sdk.description', '/docs/references/messaging/send-email.md')
+    ->label('sdk.method', 'createEmail')
+    ->label('sdk.description', '/docs/references/messaging/create-email.md')
     ->label('sdk.response.code', Response::STATUS_CODE_CREATED)
     ->label('sdk.response.type', Response::CONTENT_TYPE_JSON)
     ->label('sdk.response.model', Response::MODEL_MESSAGE)
@@ -1214,13 +1581,14 @@ App::post('/v1/messaging/messages/email')
     ->param('subject', '', new Text(998), 'Email Subject.')
     ->param('description', '', new Text(256), 'Description for Message.', true)
     ->param('content', '', new Text(64230), 'Email Content.')
-    ->param('status', 'processing', new WhiteList(['draft', 'processing']), 'Message Status.', true)
+    ->param('status', 'processing', new WhiteList(['draft', 'processing']), 'Message Status. Value must be either draft or processing.', true)
     ->param('html', false, new Boolean(), 'Is content of type HTML', true)
+    ->param('deliveryTime', null, new DatetimeValidator(requireDateInFuture: true), 'Delivery time for message in ISO 8601 format. DateTime value must be in future.', true)
     ->inject('dbForProject')
     ->inject('project')
     ->inject('messaging')
     ->inject('response')
-    ->action(function (string $messageId, string $providerId, array $to, string $subject, string $description, string $content, string $status, bool $html, Database $dbForProject, Document $project, Messaging $messaging, Response $response) {
+    ->action(function (string $messageId, string $providerId, array $to, string $subject, string $description, string $content, string $status, bool $html, ?string $deliveryTime, Database $dbForProject, Document $project, Messaging $messaging, Response $response) {
         $messageId = $messageId == 'unique()' ? ID::unique() : $messageId;
 
         $provider = $dbForProject->getDocument('providers', $providerId);
@@ -1247,8 +1615,15 @@ App::post('/v1/messaging/messages/email')
         if ($status === 'processing') {
             $messaging
                 ->setMessageId($message->getId())
-                ->setProject($project)
-                ->trigger();
+                ->setProject($project);
+
+            if (!empty($deliveryTime)) {
+                $messaging
+                    ->setDeliveryTime($deliveryTime)
+                    ->schedule();
+            } else {
+                $messaging->trigger();
+            }
         }
 
         $response
@@ -1267,7 +1642,7 @@ App::get('/v1/messaging/messages')
     ->label('sdk.response.code', Response::STATUS_CODE_OK)
     ->label('sdk.response.type', Response::CONTENT_TYPE_JSON)
     ->label('sdk.response.model', Response::MODEL_MESSAGE_LIST)
-    ->param('queries', [], new Providers(), 'Array of query strings generated using the Query class provided by the SDK. [Learn more about queries](https://appwrite.io/docs/queries). Maximum of ' . APP_LIMIT_ARRAY_PARAMS_SIZE . ' queries are allowed, each ' . APP_LIMIT_ARRAY_ELEMENT_SIZE . ' characters long. You may filter on the following attributes: ' . implode(', ', Providers::ALLOWED_ATTRIBUTES), true)
+    ->param('queries', [], new Messages(), 'Array of query strings generated using the Query class provided by the SDK. [Learn more about queries](https://appwrite.io/docs/queries). Maximum of ' . APP_LIMIT_ARRAY_PARAMS_SIZE . ' queries are allowed, each ' . APP_LIMIT_ARRAY_ELEMENT_SIZE . ' characters long. You may filter on the following attributes: ' . implode(', ', Providers::ALLOWED_ATTRIBUTES), true)
     ->inject('dbForProject')
     ->inject('response')
     ->action(function (array $queries, Database $dbForProject, Response $response) {
@@ -1279,11 +1654,9 @@ App::get('/v1/messaging/messages')
 
         if ($cursor) {
             $messageId = $cursor->getValue();
-            $cursorDocument = Authorization::skip(fn () => $dbForProject->findOne('messages', [
-                Query::equal('$id', [$messageId]),
-            ]));
+            $cursorDocument = Authorization::skip(fn () => $dbForProject->getDocument('messages', $messageId));
 
-            if ($cursorDocument === false || $cursorDocument->isEmpty()) {
+            if ($cursorDocument->isEmpty()) {
                 throw new Exception(Exception::GENERAL_CURSOR_NOT_FOUND, "Message '{$messageId}' for the 'cursor' value not found.");
             }
 
@@ -1339,20 +1712,24 @@ App::patch('/v1/messaging/messages/email/:messageId')
     ->param('subject', '', new Text(998), 'Email Subject.', true)
     ->param('description', '', new Text(256), 'Description for Message.', true)
     ->param('content', '', new Text(64230), 'Email Content.', true)
-    ->param('status', '', new WhiteList(['draft', 'processing']), 'Message Status.', true)
+    ->param('status', '', new WhiteList(['draft', 'processing']), 'Message Status. Value must be either draft or processing.', true)
     ->param('html', false, new Boolean(), 'Is content of type HTML', true)
-    ->param('deliveryTime', DateTime::now(), new DatetimeValidator(), 'Delivery time for message.', true)
+    ->param('deliveryTime', null, new DatetimeValidator(requireDateInFuture: true), 'Delivery time for message in ISO 8601 format. DateTime value must be in future.', true)
     ->inject('dbForProject')
     ->inject('project')
     ->inject('messaging')
     ->inject('response')
-    ->action(function (string $messageId, array $to, string $subject, string $description, string $content, string $status, bool $html, string $deliveryTime, Database $dbForProject, Document $project, Messaging $messaging, Response $response) {
+    ->action(function (string $messageId, array $to, string $subject, string $description, string $content, string $status, bool $html, ?string $deliveryTime, Database $dbForProject, Document $project, Messaging $messaging, Response $response) {
         $message = $dbForProject->getDocument('messages', $messageId);
 
         if ($message->isEmpty()) {
             throw new Exception(Exception::MESSAGE_NOT_FOUND);
         }
 
+        if ($message->getAttribute('status') === 'sent') {
+            throw new Exception(Exception::MESSAGE_ALREADY_SENT);
+        }
+
         if (\count($to) > 0) {
             $message->setAttribute('to', $to);
         }
@@ -1387,9 +1764,15 @@ App::patch('/v1/messaging/messages/email/:messageId')
         if ($status === 'processing') {
             $messaging
                 ->setMessageId($message->getId())
+                ->setProject($project);
+
+            if (!empty($deliveryTime)) {
+                $messaging
                 ->setDeliveryTime($deliveryTime)
-                ->setProject($project)
-                ->trigger();
+                ->schedule();
+            } else {
+                $messaging->trigger();
+            }
         }
 
         $response
diff --git a/app/init.php b/app/init.php
index 5b40ad386c..cc1c77c6ea 100644
--- a/app/init.php
+++ b/app/init.php
@@ -169,6 +169,7 @@ const DELETE_TYPE_SESSIONS = 'sessions';
 const DELETE_TYPE_CACHE_BY_TIMESTAMP = 'cacheByTimeStamp';
 const DELETE_TYPE_CACHE_BY_RESOURCE  = 'cacheByResource';
 const DELETE_TYPE_SCHEDULES = 'schedules';
+const DELETE_TYPE_SUBSCRIBERS = 'subscribers';
 // Compression type
 const COMPRESSION_TYPE_NONE = 'none';
 const COMPRESSION_TYPE_GZIP = 'gzip';
diff --git a/app/workers/deletes.php b/app/workers/deletes.php
index f831c1df3f..d5e319a662 100644
--- a/app/workers/deletes.php
+++ b/app/workers/deletes.php
@@ -127,6 +127,10 @@ class DeletesV1 extends Worker
             case DELETE_TYPE_SCHEDULES:
                 $this->deleteSchedules($this->args['datetime']);
                 break;
+            case DELETE_TYPE_SUBSCRIBERS:
+                $topic = new Document($this->args['document'] ?? []);
+                $this->deleteSubscribers($project, $topic);
+                break;
             default:
                 Console::error('No delete operation for type: ' . $type);
                 break;
@@ -170,6 +174,24 @@ class DeletesV1 extends Worker
         );
     }
 
+    /**
+     * @param Document $project
+     * @param Document $topic
+     * @throws Exception
+     */
+    protected function deleteSubscribers(Document $project, Document $topic)
+    {
+        if ($topic->isEmpty()) {
+            Console::error('Failed to delete subscribers. Topic not found');
+            return;
+        }
+        $dbForProject = $this->getProjectDB($project);
+
+        $this->deleteByGroup('subscribers', [
+            Query::equal('topicInternalId', [$topic->getInternalId()])
+        ], $dbForProject);
+    }
+
     /**
      * @param Document $project
      * @param string $resource
diff --git a/composer.json b/composer.json
index ed5cfbd825..f4d92146c0 100644
--- a/composer.json
+++ b/composer.json
@@ -43,13 +43,13 @@
         "ext-sockets": "*",
         "appwrite/php-runtimes": "0.13.*",
         "appwrite/php-clamav": "2.0.*",
-        "utopia-php/abuse": "0.31.*",
+        "utopia-php/abuse": "0.32.*",
         "utopia-php/analytics": "0.10.*",
-        "utopia-php/audit": "0.33.*",
+        "utopia-php/audit": "0.34.*",
         "utopia-php/cache": "0.8.*",
         "utopia-php/cli": "0.15.*",
         "utopia-php/config": "0.2.*",
-        "utopia-php/database": "0.43.*",
+        "utopia-php/database": "0.44.*",
         "utopia-php/domains": "0.3.*",
         "utopia-php/dsn": "0.1.*",
         "utopia-php/framework": "0.31.0",
diff --git a/composer.lock b/composer.lock
index 8abede5fa4..8f7ca86d06 100644
--- a/composer.lock
+++ b/composer.lock
@@ -4,7 +4,7 @@
         "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies",
         "This file is @generated automatically"
     ],
-    "content-hash": "34cb0b1c81424d1858df197aed030793",
+    "content-hash": "ee4518740e581a9a4889936fb584a5a4",
     "packages": [
         {
             "name": "adhocore/jwt",
@@ -156,11 +156,11 @@
         },
         {
             "name": "appwrite/php-runtimes",
-            "version": "0.13.0",
+            "version": "0.13.1",
             "source": {
                 "type": "git",
                 "url": "https://github.com/appwrite/runtimes.git",
-                "reference": "5ab496b3908992b39275994a23783701c4b3de84"
+                "reference": "b584d19cdcd82737d0ee5c34d23de791f5ed3610"
             },
             "require": {
                 "php": ">=8.0",
@@ -195,7 +195,7 @@
                 "php",
                 "runtimes"
             ],
-            "time": "2023-09-12T19:38:43+00:00"
+            "time": "2023-10-16T15:39:53+00:00"
         },
         {
             "name": "chillerlan/php-qrcode",
@@ -1861,23 +1861,23 @@
         },
         {
             "name": "utopia-php/abuse",
-            "version": "0.31.1",
+            "version": "0.32.0",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/abuse.git",
-                "reference": "b2ad372d1070f55f9545cb811b6ed2d40094e6dd"
+                "reference": "9717ffb2d7711f3fd621bb6df3edf5724c08ea78"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/abuse/zipball/b2ad372d1070f55f9545cb811b6ed2d40094e6dd",
-                "reference": "b2ad372d1070f55f9545cb811b6ed2d40094e6dd",
+                "url": "https://api.github.com/repos/utopia-php/abuse/zipball/9717ffb2d7711f3fd621bb6df3edf5724c08ea78",
+                "reference": "9717ffb2d7711f3fd621bb6df3edf5724c08ea78",
                 "shasum": ""
             },
             "require": {
                 "ext-curl": "*",
                 "ext-pdo": "*",
                 "php": ">=8.0",
-                "utopia-php/database": "0.43.*"
+                "utopia-php/database": "0.44.*"
             },
             "require-dev": {
                 "laravel/pint": "1.5.*",
@@ -1904,9 +1904,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/abuse/issues",
-                "source": "https://github.com/utopia-php/abuse/tree/0.31.1"
+                "source": "https://github.com/utopia-php/abuse/tree/0.32.0"
             },
-            "time": "2023-08-29T11:07:46+00:00"
+            "time": "2023-10-18T07:28:55+00:00"
         },
         {
             "name": "utopia-php/analytics",
@@ -1956,21 +1956,21 @@
         },
         {
             "name": "utopia-php/audit",
-            "version": "0.33.1",
+            "version": "0.34.0",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/audit.git",
-                "reference": "c117e8e9ce4e3e1b369e8b5b55b2d6ab3138eadd"
+                "reference": "cf34cc3f9f20da4e574a9be4517e1a11025a858f"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/audit/zipball/c117e8e9ce4e3e1b369e8b5b55b2d6ab3138eadd",
-                "reference": "c117e8e9ce4e3e1b369e8b5b55b2d6ab3138eadd",
+                "url": "https://api.github.com/repos/utopia-php/audit/zipball/cf34cc3f9f20da4e574a9be4517e1a11025a858f",
+                "reference": "cf34cc3f9f20da4e574a9be4517e1a11025a858f",
                 "shasum": ""
             },
             "require": {
                 "php": ">=8.0",
-                "utopia-php/database": "0.43.*"
+                "utopia-php/database": "0.44.*"
             },
             "require-dev": {
                 "laravel/pint": "1.5.*",
@@ -1997,9 +1997,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/audit/issues",
-                "source": "https://github.com/utopia-php/audit/tree/0.33.1"
+                "source": "https://github.com/utopia-php/audit/tree/0.34.0"
             },
-            "time": "2023-08-29T11:07:40+00:00"
+            "time": "2023-10-18T07:43:25+00:00"
         },
         {
             "name": "utopia-php/cache",
@@ -2152,16 +2152,16 @@
         },
         {
             "name": "utopia-php/database",
-            "version": "0.43.5",
+            "version": "0.44.1",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/database.git",
-                "reference": "5f7b05189cfbcc0506090498c580c5765375a00a"
+                "reference": "e0b832d217e4d429c96ade671e85ece942446543"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/database/zipball/5f7b05189cfbcc0506090498c580c5765375a00a",
-                "reference": "5f7b05189cfbcc0506090498c580c5765375a00a",
+                "url": "https://api.github.com/repos/utopia-php/database/zipball/e0b832d217e4d429c96ade671e85ece942446543",
+                "reference": "e0b832d217e4d429c96ade671e85ece942446543",
                 "shasum": ""
             },
             "require": {
@@ -2202,9 +2202,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/database/issues",
-                "source": "https://github.com/utopia-php/database/tree/0.43.5"
+                "source": "https://github.com/utopia-php/database/tree/0.44.1"
             },
-            "time": "2023-10-06T06:49:47+00:00"
+            "time": "2023-10-18T07:05:41+00:00"
         },
         {
             "name": "utopia-php/domains",
diff --git a/docker-compose.yml b/docker-compose.yml
index f0ddbfe41b..a365ce17ba 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -192,11 +192,11 @@ services:
       - _APP_MESSAGE_SMS_PROVIDER_MSG91_AUTH_KEY
       - _APP_MESSAGE_SMS_PROVIDER_MSG91_FROM
       - _APP_MESSAGE_SMS_PROVIDER_MSG91_TO
-      - _APP_MESSAGE_SMS_PROVIDER_MAILGUN_API_KEY
-      - _APP_MESSAGE_SMS_PROVIDER_MAILGUN_DOMAIN
-      - _APP_MESSAGE_SMS_PROVIDER_MAILGUN_FROM
-      - _APP_MESSAGE_SMS_PROVIDER_MAILGUN_RECEIVER_EMAIL
-      - _APP_MESSAGE_SMS_PROVIDER_MAILGUN_IS_EU_REGION
+      - _APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_API_KEY
+      - _APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_DOMAIN
+      - _APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_FROM
+      - _APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_RECEIVER_EMAIL
+      - _APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_IS_EU_REGION
 
   appwrite-realtime:
     entrypoint: realtime
diff --git a/src/Appwrite/Event/Messaging.php b/src/Appwrite/Event/Messaging.php
index 18887ffcd3..45ef843caa 100644
--- a/src/Appwrite/Event/Messaging.php
+++ b/src/Appwrite/Event/Messaging.php
@@ -2,8 +2,9 @@
 
 namespace Appwrite\Event;
 
+use Resque;
 use ResqueScheduler;
-use Utopia\Database\DateTime;
+use Utopia\Database\Document;
 
 class Messaging extends Event
 {
@@ -61,16 +62,46 @@ class Messaging extends Event
         return $this->deliveryTime;
     }
 
+    /**
+     * Set project for this event.
+     *
+     * @param Document $project
+     * @return self
+     */
+    public function setProject(Document $project): self
+    {
+        $this->project = $project;
+
+        return $this;
+    }
+
     /**
      * Executes the event and sends it to the messaging worker.
+     * @return string|bool
+     * @throws \InvalidArgumentException
      */
     public function trigger(): string | bool
     {
-        ResqueScheduler::enqueueAt(!empty($this->deliveryTime) ? $this->deliveryTime : DateTime::now(), $this->queue, $this->class, [
+        return Resque::enqueue($this->queue, $this->class, [
+            'project' => $this->project,
+            'user' => $this->user,
+            'messageId' => $this->messageId,
+        ]);
+    }
+
+    /**
+     * Schedules the messaging event and schedules it in the messaging worker queue.
+     *
+     * @return void
+     * @throws \Resque_Exception
+     * @throws \ResqueScheduler_InvalidTimestampException
+     */
+    public function schedule(): void
+    {
+        ResqueScheduler::enqueueAt(new \DateTime($this->deliveryTime, new \DateTimeZone('UTC')), $this->queue, $this->class, [
             'project' => $this->project,
             'user' => $this->user,
             'messageId' => $this->messageId,
         ]);
-        return true;
     }
 }
diff --git a/src/Appwrite/Extend/Exception.php b/src/Appwrite/Extend/Exception.php
index cdd2538abb..ee17ccef0c 100644
--- a/src/Appwrite/Extend/Exception.php
+++ b/src/Appwrite/Extend/Exception.php
@@ -230,8 +230,17 @@ class Exception extends \Exception
     public const PROVIDER_ALREADY_EXISTS            = 'provider_already_exists';
     public const PROVIDER_INCORRECT_TYPE            = 'provider_incorrect_type';
 
+    /** Topic */
+    public const TOPIC_NOT_FOUND                    = 'topic_not_found';
+    public const TOPIC_ALREADY_EXISTS               = 'topic_already_exists';
+
+    /** Subscriber */
+    public const SUBSCRIBER_NOT_FOUND               = 'subscriber_not_found';
+    public const SUBSCRIBER_ALREADY_EXISTS          = 'subscriber_already_exists';
+
     /** Message */
     public const MESSAGE_NOT_FOUND                  = 'message_not_found';
+    public const MESSAGE_ALREADY_SENT               = 'message_already_sent';
 
     protected $type = '';
     protected $errors = [];
diff --git a/src/Appwrite/Utopia/Database/Validator/Queries/Messages.php b/src/Appwrite/Utopia/Database/Validator/Queries/Messages.php
new file mode 100644
index 0000000000..d21aa5de1b
--- /dev/null
+++ b/src/Appwrite/Utopia/Database/Validator/Queries/Messages.php
@@ -0,0 +1,27 @@
+<?php
+
+namespace Appwrite\Utopia\Database\Validator\Queries;
+
+class Messages extends Base
+{
+    public const ALLOWED_ATTRIBUTES = [
+        'to',
+        'providerId',
+        'deliveredAt',
+        'deliveredTo',
+        'deliveryErrors',
+        'status',
+        'description',
+        'data',
+        'search'
+    ];
+
+    /**
+     * Expression constructor
+     *
+     */
+    public function __construct()
+    {
+        parent::__construct('messages', self::ALLOWED_ATTRIBUTES);
+    }
+}
diff --git a/src/Appwrite/Utopia/Database/Validator/Queries/Providers.php b/src/Appwrite/Utopia/Database/Validator/Queries/Providers.php
index e72153734c..1fd6c9e9f8 100644
--- a/src/Appwrite/Utopia/Database/Validator/Queries/Providers.php
+++ b/src/Appwrite/Utopia/Database/Validator/Queries/Providers.php
@@ -9,7 +9,7 @@ class Providers extends Base
         'provider',
         'type',
         'default',
-        'enabled'
+        'enabled',
     ];
 
     /**
diff --git a/src/Appwrite/Utopia/Database/Validator/Queries/Subscribers.php b/src/Appwrite/Utopia/Database/Validator/Queries/Subscribers.php
new file mode 100644
index 0000000000..55bb455903
--- /dev/null
+++ b/src/Appwrite/Utopia/Database/Validator/Queries/Subscribers.php
@@ -0,0 +1,20 @@
+<?php
+
+namespace Appwrite\Utopia\Database\Validator\Queries;
+
+class Subscribers extends Base
+{
+    public const ALLOWED_ATTRIBUTES = [
+        'targetId',
+        'topicId'
+    ];
+
+    /**
+     * Expression constructor
+     *
+     */
+    public function __construct()
+    {
+        parent::__construct('subscribers', self::ALLOWED_ATTRIBUTES);
+    }
+}
diff --git a/src/Appwrite/Utopia/Database/Validator/Queries/Topics.php b/src/Appwrite/Utopia/Database/Validator/Queries/Topics.php
new file mode 100644
index 0000000000..120b3edc3f
--- /dev/null
+++ b/src/Appwrite/Utopia/Database/Validator/Queries/Topics.php
@@ -0,0 +1,20 @@
+<?php
+
+namespace Appwrite\Utopia\Database\Validator\Queries;
+
+class Topics extends Base
+{
+    public const ALLOWED_ATTRIBUTES = [
+        'name',
+        'description',
+    ];
+
+    /**
+     * Expression constructor
+     *
+     */
+    public function __construct()
+    {
+        parent::__construct('topics', self::ALLOWED_ATTRIBUTES);
+    }
+}
diff --git a/src/Appwrite/Utopia/Response/Model/Message.php b/src/Appwrite/Utopia/Response/Model/Message.php
index 5e4a358490..0cadc378d9 100644
--- a/src/Appwrite/Utopia/Response/Model/Message.php
+++ b/src/Appwrite/Utopia/Response/Model/Message.php
@@ -58,6 +58,15 @@ class Message extends Any
                 'default' => 0,
                 'example' => 1,
             ])
+            ->addRule('data', [
+                'type' => self::TYPE_JSON,
+                'description' => 'Data of the message.',
+                'default' => [],
+                'example' => [
+                    'subject' => 'Welcome to Appwrite',
+                    'content' => 'Hi there, welcome to Appwrite family.',
+                ],
+            ])
             ->addRule('status', [
                 'type' => self::TYPE_STRING,
                 'description' => 'Status of delivery.',
diff --git a/tests/e2e/Scopes/ProjectCustom.php b/tests/e2e/Scopes/ProjectCustom.php
index 5ab199f2bb..7fdd96e5be 100644
--- a/tests/e2e/Scopes/ProjectCustom.php
+++ b/tests/e2e/Scopes/ProjectCustom.php
@@ -87,6 +87,10 @@ trait ProjectCustom
                 'providers.write',
                 'messages.read',
                 'messages.write',
+                'topics.write',
+                'topics.read',
+                'subscribers.write',
+                'subscribers.read'
             ],
         ]);
 
diff --git a/tests/e2e/Services/GraphQL/Base.php b/tests/e2e/Services/GraphQL/Base.php
index 0a93ac34c4..f3947c53b4 100644
--- a/tests/e2e/Services/GraphQL/Base.php
+++ b/tests/e2e/Services/GraphQL/Base.php
@@ -107,6 +107,11 @@ trait Base
     public static string $DELETE_USER_SESSIONS = 'delete_user_sessions';
     public static string $DELETE_USER_SESSION = 'delete_user_session';
     public static string $DELETE_USER = 'delete_user';
+    public static string $CREATE_USER_TARGET = 'create_user_target';
+    public static string $LIST_USER_TARGETS = 'list_user_targets';
+    public static string $GET_USER_TARGET = 'get_user_target';
+    public static string $UPDATE_USER_TARGET = 'update_user_target';
+    public static string $DELETE_USER_TARGET = 'delete_user_target';
 
     // Teams
     public static string $GET_TEAM = 'get_team';
@@ -220,6 +225,24 @@ trait Base
     public static string $UPDATE_APNS_PROVIDER = 'update_apns_provider';
     public static string $DELETE_PROVIDER = 'delete_provider';
 
+    // Topics
+    public static string $CREATE_TOPIC = 'create_topic';
+    public static string $LIST_TOPICS = 'list_topics';
+    public static string $GET_TOPIC = 'get_topic';
+    public static string $UPDATE_TOPIC = 'update_topic';
+    public static string $DELETE_TOPIC = 'delete_topic';
+
+    // Subscriptions
+    public static string $CREATE_SUBSCRIBER = 'create_subscriber';
+    public static string $LIST_SUBSCRIBERS = 'list_subscribers';
+    public static string $GET_SUBSCRIBER = 'get_subscriber';
+    public static string $DELETE_SUBSCRIBER = 'delete_subscriber';
+
+    // Messages
+    public static string $CREATE_EMAIL = 'create_email';
+    public static string $LIST_MESSAGES = 'list_messages';
+    public static string $GET_MESSAGE = 'get_message';
+
     // Complex queries
     public static string $COMPLEX_QUERY = 'complex_query';
 
@@ -902,6 +925,51 @@ trait Base
                         status
                     }
                 }';
+            case self::$CREATE_USER_TARGET:
+                return 'mutation createUserTarget($userId: String!, $targetId: String!, $providerId: String!, $identifier: String!){
+                    usersCreateTarget(userId: $userId, targetId: $targetId, providerId: $providerId, identifier: $identifier) {
+                        _id
+                        userId
+                        providerId
+                        identifier
+                    }
+                }';
+            case self::$LIST_USER_TARGETS:
+                return 'query listUserTargets($userId: String!) {
+                    usersListTargets(userId: $userId) {
+                        total
+                        targets {
+                            _id
+                            userId
+                            providerId
+                            identifier
+                        }
+                    }
+                }';
+            case self::$GET_USER_TARGET:
+                return 'query getUserTarget($userId: String!, $targetId: String!) {
+                    usersGetTarget(userId: $userId, targetId: $targetId) {
+                        _id
+                        userId
+                        providerId
+                        identifier
+                    }
+                }';
+            case self::$UPDATE_USER_TARGET:
+                return 'mutation updateUserTarget($userId: String!, $targetId: String!, $identifier: String!){
+                    usersUpdateTargetIdentifier(userId: $userId, targetId: $targetId, identifier: $identifier) {
+                        _id
+                        userId
+                        providerId
+                        identifier
+                    }
+                }';
+            case self::$DELETE_USER_TARGET:
+                return 'mutation deleteUserTarget($userId: String!, $targetId: String!){
+                    usersDeleteTarget(userId: $userId, targetId: $targetId) {
+                        status
+                    }
+                }';
             case self::$GET_LOCALE:
                 return 'query getLocale {
                     localeGet {
@@ -1938,6 +2006,129 @@ trait Base
                         status
                     }
                 }';
+            case self::$CREATE_TOPIC:
+                return 'mutation createTopic($providerId: String!, $topicId: String!, $name: String!, $description: String!) {
+                    messagingCreateTopic(providerId: $providerId, topicId: $topicId, name: $name, description: $description) {
+                        _id
+                        name
+                        providerId
+                        description
+                    }
+                }';
+            case self::$LIST_TOPICS:
+                return 'query listTopics {
+                    messagingListTopics {
+                        total
+                        topics {
+                            _id
+                            name
+                            providerId
+                            description
+                        }
+                    }
+                }';
+            case self::$GET_TOPIC:
+                return 'query getTopic($topicId: String!) {
+                    messagingGetTopic(topicId: $topicId) {
+                        _id
+                        name
+                        providerId
+                        description
+                    }
+                }';
+            case self::$UPDATE_TOPIC:
+                return 'mutation updateTopic($topicId: String!, $name: String!, $description: String!) {
+                    messagingUpdateTopic(topicId: $topicId, name: $name, description: $description) {
+                        _id
+                        name
+                        providerId
+                        description
+                    }
+                }';
+            case self::$DELETE_TOPIC:
+                return 'mutation deleteTopic($topicId: String!) {
+                    messagingDeleteTopic(topicId: $topicId) {
+                        status
+                    }
+                }';
+            case self::$CREATE_SUBSCRIBER:
+                return 'mutation createSubscriber($subscriberId: String!, $targetId: String!, $topicId: String!) {
+                    messagingCreateSubscriber(subscriberId: $subscriberId, targetId: $targetId, topicId: $topicId) {
+                        _id
+                        targetId
+                        topicId
+                    }
+                }';
+            case self::$LIST_SUBSCRIBERS:
+                return 'query listSubscribers($topicId: String!) {
+                    messagingListSubscribers(topicId: $topicId) {
+                        total
+                        subscribers {
+                            _id
+                            targetId
+                            topicId
+                        }
+                    }
+                }';
+            case self::$GET_SUBSCRIBER:
+                return 'query getSubscriber($topicId: String!, $subscriberId: String!) {
+                    messagingGetSubscriber(topicId: $topicId, subscriberId: $subscriberId) {
+                        _id
+                        targetId
+                        topicId
+                    }
+                }';
+            case self::$DELETE_SUBSCRIBER:
+                return 'mutation deleteSubscriber($topicId: String!, $subscriberId: String!) {
+                    messagingDeleteSubscriber(topicId: $topicId, subscriberId: $subscriberId) {
+                        status
+                    }
+            }';
+            case self::$CREATE_EMAIL:
+                return 'mutation createEmail($messageId: String!, $providerId: String!, $to: [String!]!, $subject: String!, $content: String!, $status: String, $description: String, $html: Boolean, $deliveryTime: String) {
+                    messagingCreateEmail(messageId: $messageId, providerId: $providerId, to: $to, subject: $subject, content: $content, status: $status, description: $description, html: $html, deliveryTime: $deliveryTime) {
+                        _id
+                        providerId
+                        to
+                        deliveryTime
+                        deliveredAt
+                        deliveryErrors
+                        deliveredTo
+                        status
+                        description
+                    }
+                }';
+            case self::$LIST_MESSAGES:
+                return 'query listMessages {
+                    messagingListMessages {
+                        total
+                        messages {
+                            _id
+                            providerId
+                            to
+                            deliveryTime
+                            deliveredAt
+                            deliveryErrors
+                            deliveredTo
+                            status
+                            description
+                        }
+                    }
+                }';
+            case self::$GET_MESSAGE:
+                return 'query getMessage($messageId: String!) {
+                    messagingGetMessage(messageId: $messageId) {
+                        _id
+                        providerId
+                        to
+                        deliveryTime
+                        deliveredAt
+                        deliveryErrors
+                        deliveredTo
+                        status
+                        description
+                    }
+                }';
             case self::$COMPLEX_QUERY:
                 return 'mutation complex($databaseId: String!, $databaseName: String!, $collectionId: String!, $collectionName: String!, $documentSecurity: Boolean!, $collectionPermissions: [String!]!) {
                     databasesCreate(databaseId: $databaseId, name: $databaseName) {
diff --git a/tests/e2e/Services/GraphQL/MessagingTest.php b/tests/e2e/Services/GraphQL/MessagingTest.php
index 99bad52887..8c3e4970e7 100644
--- a/tests/e2e/Services/GraphQL/MessagingTest.php
+++ b/tests/e2e/Services/GraphQL/MessagingTest.php
@@ -6,6 +6,7 @@ use Tests\E2E\Client;
 use Tests\E2E\Scopes\ProjectCustom;
 use Tests\E2E\Scopes\Scope;
 use Tests\E2E\Scopes\SideServer;
+use Utopia\App;
 use Utopia\Database\Helpers\ID;
 
 class MessagingTest extends Scope
@@ -257,4 +258,405 @@ class MessagingTest extends Scope
             $this->assertEquals(204, $response['headers']['status-code']);
         }
     }
+
+    public function testCreateTopic()
+    {
+        $providerParam = [
+            'sendgrid' => [
+                'providerId' => ID::unique(),
+                'name' => 'Sengrid1',
+                'apiKey' => 'my-apikey',
+            ]
+        ];
+        $query = $this->getQuery(self::$CREATE_SENDGRID_PROVIDER);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => $providerParam['sendgrid'],
+        ];
+        $response = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $providerId = $response['body']['data']['messagingCreateSendgridProvider']['_id'];
+
+        $query = $this->getQuery(self::$CREATE_TOPIC);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'providerId' => $providerId,
+                'topicId' => ID::unique(),
+                'name' => 'topic1',
+                'description' => 'Active users',
+            ],
+        ];
+        $response = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(200, $response['headers']['status-code']);
+        $this->assertEquals('topic1', $response['body']['data']['messagingCreateTopic']['name']);
+        $this->assertEquals('Active users', $response['body']['data']['messagingCreateTopic']['description']);
+
+        return $response['body']['data']['messagingCreateTopic'];
+    }
+
+    /**
+     * @depends testCreateTopic
+     */
+    public function testUpdateTopic(array $topic)
+    {
+        $topicId = $topic['_id'];
+        $query = $this->getQuery(self::$UPDATE_TOPIC);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'topicId' => $topicId,
+                'name' => 'topic2',
+                'description' => 'Inactive users',
+            ],
+        ];
+        $response = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(200, $response['headers']['status-code']);
+        $this->assertEquals('topic2', $response['body']['data']['messagingUpdateTopic']['name']);
+        $this->assertEquals('Inactive users', $response['body']['data']['messagingUpdateTopic']['description']);
+
+        return $topicId;
+    }
+
+    /**
+     * @depends testCreateTopic
+     */
+    public function testListTopics()
+    {
+        $query = $this->getQuery(self::$LIST_TOPICS);
+        $graphQLPayload = [
+            'query' => $query,
+        ];
+        $response = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(200, $response['headers']['status-code']);
+        $this->assertEquals(1, \count($response['body']['data']['messagingListTopics']['topics']));
+    }
+
+    /**
+     * @depends testUpdateTopic
+     */
+    public function testGetTopic(string $topicId)
+    {
+        $query = $this->getQuery(self::$GET_TOPIC);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'topicId' => $topicId,
+            ],
+        ];
+        $response = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(200, $response['headers']['status-code']);
+        $this->assertEquals('topic2', $response['body']['data']['messagingGetTopic']['name']);
+        $this->assertEquals('Inactive users', $response['body']['data']['messagingGetTopic']['description']);
+    }
+
+    /**
+     * @depends testCreateTopic
+     */
+    public function testCreateSubscriber(array $topic)
+    {
+        $topicId = $topic['_id'];
+
+        $userId = $this->getUser()['$id'];
+
+        $query = $this->getQuery(self::$CREATE_USER_TARGET);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'targetId' => ID::unique(),
+                'userId' => $userId,
+                'providerId' => $topic['providerId'],
+                'identifier' => 'token',
+            ],
+        ];
+        $response = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(200, $response['headers']['status-code']);
+        $this->assertEquals($userId, $response['body']['data']['usersCreateTarget']['userId']);
+        $this->assertEquals('token', $response['body']['data']['usersCreateTarget']['identifier']);
+
+        $targetId = $response['body']['data']['usersCreateTarget']['_id'];
+
+        $query = $this->getQuery(self::$CREATE_SUBSCRIBER);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'subscriberId' => ID::unique(),
+                'topicId' => $topicId,
+                'targetId' => $targetId,
+            ],
+        ];
+        $response = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+        ], $this->getHeaders()), $graphQLPayload);
+
+        $this->assertEquals(200, $response['headers']['status-code']);
+
+        return $response['body']['data']['messagingCreateSubscriber'];
+    }
+
+    /**
+     * @depends testUpdateTopic
+     */
+    public function testListSubscribers(string $topicId)
+    {
+        $query = $this->getQuery(self::$LIST_SUBSCRIBERS);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'topicId' => $topicId,
+            ],
+        ];
+        $response = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(200, $response['headers']['status-code']);
+        $this->assertEquals(1, \count($response['body']['data']['messagingListSubscribers']['subscribers']));
+    }
+
+    /**
+     * @depends testCreateSubscriber
+     */
+    public function testGetSubscriber(array $subscriber)
+    {
+        $topicId = $subscriber['topicId'];
+        $subscriberId = $subscriber['_id'];
+
+        $query = $this->getQuery(self::$GET_SUBSCRIBER);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'topicId' => $topicId,
+                'subscriberId' => $subscriberId,
+            ],
+        ];
+
+        $response = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(200, $response['headers']['status-code']);
+        $this->assertEquals($subscriberId, $response['body']['data']['messagingGetSubscriber']['_id']);
+    }
+
+    /**
+     * @depends testCreateSubscriber
+     */
+    public function testDeleteSubscriber(array $subscriber)
+    {
+        $topicId = $subscriber['topicId'];
+        $subscriberId = $subscriber['_id'];
+
+        $query = $this->getQuery(self::$DELETE_SUBSCRIBER);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'topicId' => $topicId,
+                'subscriberId' => $subscriberId,
+            ],
+        ];
+
+        $response = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+        ], $this->getHeaders()), $graphQLPayload);
+
+        $this->assertEquals(200, $response['headers']['status-code']);
+    }
+
+    /**
+     * @depends testUpdateTopic
+     */
+    public function testDeleteTopic(string $topicId)
+    {
+        $query = $this->getQuery(self::$DELETE_TOPIC);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'topicId' => $topicId,
+            ],
+        ];
+        $response = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(204, $response['headers']['status-code']);
+    }
+
+    public function testSendEmail()
+    {
+        $to = App::getEnv('_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_RECEIVER_EMAIL');
+        $from = App::getEnv('_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_FROM');
+        $apiKey = App::getEnv('_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_API_KEY');
+        $domain = App::getEnv('_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_DOMAIN');
+        $isEuRegion = App::getEnv('_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_IS_EU_REGION');
+        if (empty($to) || empty($from) || empty($apiKey) || empty($domain) || empty($isEuRegion)) {
+            $this->markTestSkipped('Email provider not configured');
+        }
+
+        $query = $this->getQuery(self::$CREATE_MAILGUN_PROVIDER);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'providerId' => ID::unique(),
+                'name' => 'Mailgun1',
+                'apiKey' => $apiKey,
+                'domain' => $domain,
+                'from' => $from,
+                'isEuRegion' => $isEuRegion,
+            ],
+        ];
+        $provider = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(200, $provider['headers']['status-code']);
+
+        $providerId = $provider['body']['data']['messagingCreateMailgunProvider']['_id'];
+
+        $query = $this->getQuery(self::$CREATE_TOPIC);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'providerId' => $providerId,
+                'topicId' => ID::unique(),
+                'name' => 'topic1',
+                'description' => 'Active users',
+            ],
+        ];
+        $topic = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(200, $topic['headers']['status-code']);
+
+        $query = $this->getQuery(self::$CREATE_USER);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'userId' => ID::custom('test-user'),
+                'email' => $to,
+                'password' => 'password',
+                'name' => 'Messaging User',
+            ]
+        ];
+        $user = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(200, $user['headers']['status-code']);
+
+        $query = $this->getQuery(self::$CREATE_USER_TARGET);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'targetId' => ID::unique(),
+                'userId' => $user['body']['data']['usersCreate']['_id'],
+                'providerId' => $providerId,
+                'identifier' => $to,
+            ],
+        ];
+        $target = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(200, $target['headers']['status-code']);
+
+        $query = $this->getQuery(self::$CREATE_SUBSCRIBER);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'subscriberId' => ID::unique(),
+                'topicId' => $topic['body']['data']['messagingCreateTopic']['_id'],
+                'targetId' => $target['body']['data']['usersCreateTarget']['_id'],
+            ],
+        ];
+        $subscriber = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+        ], $this->getHeaders()), $graphQLPayload);
+
+        $this->assertEquals(200, $subscriber['headers']['status-code']);
+
+        $query = $this->getQuery(self::$CREATE_EMAIL);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'messageId' => ID::unique(),
+                'providerId' => $providerId,
+                'to' => [$topic['body']['data']['messagingCreateTopic']['_id']],
+                'subject' => 'Khali beats Undertaker',
+                'content' => 'https://www.youtube.com/watch?v=dQw4w9WgXcQ',
+            ],
+        ];
+        $email = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(200, $email['headers']['status-code']);
+
+        \sleep(5);
+
+        $query = $this->getQuery(self::$GET_MESSAGE);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'messageId' => $email['body']['data']['messagingCreateEmail']['_id'],
+            ],
+        ];
+        $message = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), $graphQLPayload);
+
+        $this->assertEquals(200, $message['headers']['status-code']);
+        $this->assertEquals(1, $message['body']['data']['messagingGetMessage']['deliveredTo']);
+        $this->assertEquals(0, \count($message['body']['data']['messagingGetMessage']['deliveryErrors']));
+    }
 }
diff --git a/tests/e2e/Services/GraphQL/UsersTest.php b/tests/e2e/Services/GraphQL/UsersTest.php
index 9bd503df0f..c0327381cd 100644
--- a/tests/e2e/Services/GraphQL/UsersTest.php
+++ b/tests/e2e/Services/GraphQL/UsersTest.php
@@ -45,6 +45,55 @@ class UsersTest extends Scope
         return $user;
     }
 
+    /**
+     * @depends testCreateUser
+     */
+    public function testCreateUserTarget(array $user)
+    {
+        $projectId = $this->getProject()['$id'];
+
+        $query = $this->getQuery(self::$CREATE_MAILGUN_PROVIDER);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'providerId' => ID::unique(),
+                'name' => 'Mailgun1',
+                'apiKey' => 'api-key',
+                'domain' => 'domain',
+                'from' => 'from@domain',
+                'isEuRegion' => false,
+            ],
+        ];
+        $provider = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ], $this->getHeaders()), $graphQLPayload);
+        $providerId = $provider['body']['data']['messagingCreateMailgunProvider']['_id'];
+
+        $this->assertEquals(200, $provider['headers']['status-code']);
+
+        $query = $this->getQuery(self::$CREATE_USER_TARGET);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'targetId' => ID::unique(),
+                'userId' => $user['_id'],
+                'providerId' => $providerId,
+                'identifier' => 'identifier',
+            ]
+        ];
+
+        $target = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ], $this->getHeaders()), $graphQLPayload);
+
+        $this->assertEquals(200, $target['headers']['status-code']);
+        $this->assertEquals('identifier', $target['body']['data']['usersCreateTarget']['identifier']);
+
+        return $target['body']['data']['usersCreateTarget'];
+    }
+
     public function testGetUsers()
     {
         $projectId = $this->getProject()['$id'];
@@ -176,6 +225,54 @@ class UsersTest extends Scope
         $this->assertIsArray($user['body']['data']['usersListLogs']);
     }
 
+    /**
+     * @depends testCreateUserTarget
+     */
+    public function testListUserTargets(array $target)
+    {
+        $projectId = $this->getProject()['$id'];
+        $query = $this->getQuery(self::$LIST_USER_TARGETS);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'userId' => $target['userId'],
+            ]
+        ];
+
+        $targets = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ], $this->getHeaders()), $graphQLPayload);
+
+        $this->assertEquals(200, $targets['headers']['status-code']);
+        $this->assertIsArray($targets['body']['data']['usersListTargets']);
+        $this->assertCount(1, $targets['body']['data']['usersListTargets']['targets']);
+    }
+
+    /**
+     * @depends testCreateUserTarget
+     */
+    public function testGetUserTarget(array $target)
+    {
+        $projectId = $this->getProject()['$id'];
+        $query = $this->getQuery(self::$GET_USER_TARGET);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'userId' => $target['userId'],
+                'targetId' => $target['_id'],
+            ]
+        ];
+
+        $target = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ], $this->getHeaders()), $graphQLPayload);
+
+        $this->assertEquals(200, $target['headers']['status-code']);
+        $this->assertEquals('identifier', $target['body']['data']['usersGetTarget']['identifier']);
+    }
+
     public function testUpdateUserStatus()
     {
         $projectId = $this->getProject()['$id'];
@@ -360,6 +457,31 @@ class UsersTest extends Scope
         $this->assertEquals('{"key":"value"}', $user['body']['data']['usersUpdatePrefs']['data']);
     }
 
+    /**
+     * @depends testCreateUserTarget
+     */
+    public function testUpdateUserTarget(array $target)
+    {
+        $projectId = $this->getProject()['$id'];
+        $query = $this->getQuery(self::$UPDATE_USER_TARGET);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'userId' => $target['userId'],
+                'targetId' => $target['_id'],
+                'identifier' => 'newidentifier',
+            ],
+        ];
+
+        $target = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ], $this->getHeaders()), $graphQLPayload);
+
+        $this->assertEquals(200, $target['headers']['status-code']);
+        $this->assertEquals('newidentifier', $target['body']['data']['usersUpdateTargetIdentifier']['identifier']);
+    }
+
     public function testDeleteUserSessions()
     {
         $projectId = $this->getProject()['$id'];
@@ -407,6 +529,29 @@ class UsersTest extends Scope
         $this->getUser();
     }
 
+    /**
+     * @depends testCreateUserTarget
+     */
+    public function testDeleteUserTarget(array $target)
+    {
+        $projectId = $this->getProject()['$id'];
+        $query = $this->getQuery(self::$DELETE_USER_TARGET);
+        $graphQLPayload = [
+            'query' => $query,
+            'variables' => [
+                'userId' => $target['userId'],
+                'targetId' => $target['_id'],
+            ]
+        ];
+
+        $target = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $projectId,
+        ], $this->getHeaders()), $graphQLPayload);
+
+        $this->assertEquals(204, $target['headers']['status-code']);
+    }
+
     public function testDeleteUser()
     {
         $projectId = $this->getProject()['$id'];
diff --git a/tests/e2e/Services/Messaging/MessagingBase.php b/tests/e2e/Services/Messaging/MessagingBase.php
index 3282330d86..1c349a2dbe 100644
--- a/tests/e2e/Services/Messaging/MessagingBase.php
+++ b/tests/e2e/Services/Messaging/MessagingBase.php
@@ -3,6 +3,7 @@
 namespace Tests\E2E\Services\Messaging;
 
 use Tests\E2E\Client;
+use Utopia\App;
 use Utopia\Database\Helpers\ID;
 
 trait MessagingBase
@@ -208,4 +209,271 @@ trait MessagingBase
             $this->assertEquals(204, $response['headers']['status-code']);
         }
     }
+
+    public function testCreateTopic(): array
+    {
+        $provider = $this->client->call(Client::METHOD_POST, '/messaging/providers/sendgrid', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), [
+            'providerId' => 'unique()',
+            'name' => 'Sendgrid1',
+            'apiKey' => 'my-apikey',
+        ]);
+        $this->assertEquals(201, $provider['headers']['status-code']);
+        $response = $this->client->call(Client::METHOD_POST, '/messaging/topics', [
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ], [
+            'providerId' => $provider['body']['$id'],
+            'topicId' => 'unique()',
+            'name' => 'my-app',
+            'description' => 'web app'
+        ]);
+        $this->assertEquals(201, $response['headers']['status-code']);
+        $this->assertEquals('my-app', $response['body']['name']);
+
+        return $response['body'];
+    }
+
+    /**
+     * @depends testCreateTopic
+     */
+    public function testUpdateTopic(array $topic): string
+    {
+        $response = $this->client->call(Client::METHOD_PATCH, '/messaging/topics/' . $topic['$id'], [
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ], [
+            'name' => 'android-app',
+            'description' => 'updated-description'
+        ]);
+        $this->assertEquals(200, $response['headers']['status-code']);
+        $this->assertEquals('android-app', $response['body']['name']);
+        $this->assertEquals('updated-description', $response['body']['description']);
+        return $response['body']['$id'];
+    }
+
+    public function testListTopic()
+    {
+        $response = $this->client->call(Client::METHOD_GET, '/messaging/topics', [
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]);
+        $this->assertEquals(200, $response['headers']['status-code']);
+        $this->assertEquals(1, \count($response['body']['topics']));
+    }
+
+    /**
+     * @depends testUpdateTopic
+     */
+    public function testGetTopic(string $topicId)
+    {
+        $response = $this->client->call(Client::METHOD_GET, '/messaging/topics/' . $topicId, [
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]);
+        $this->assertEquals(200, $response['headers']['status-code']);
+        $this->assertEquals('android-app', $response['body']['name']);
+        $this->assertEquals('updated-description', $response['body']['description']);
+    }
+
+    /**
+     * @depends testCreateTopic
+     */
+    public function testCreateSubscriber(array $topic)
+    {
+        $userId = $this->getUser()['$id'];
+        $target = $this->client->call(Client::METHOD_POST, '/users/' . $userId . '/targets', array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), [
+            'targetId' => ID::unique(),
+            'providerId' => $topic['providerId'],
+            'identifier' => 'my-token',
+        ]);
+        $this->assertEquals(201, $target['headers']['status-code']);
+
+        $response = $this->client->call(Client::METHOD_POST, '/messaging/topics/' . $topic['$id'] . '/subscribers', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+        ], $this->getHeaders()), [
+            'subscriberId' => 'unique()',
+            'targetId' => $target['body']['$id'],
+        ]);
+        $this->assertEquals(201, $response['headers']['status-code']);
+        return [
+            'topicId' => $topic['$id'],
+            'targetId' => $target['body']['$id'],
+            'subscriberId' => $response['body']['$id']
+        ];
+    }
+
+    /**
+     * @depends testCreateSubscriber
+     */
+    public function testGetSubscriber(array $data)
+    {
+        $response = $this->client->call(Client::METHOD_GET, '/messaging/topics/' . $data['topicId'] . '/subscriber/' . $data['subscriberId'], \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]));
+        $this->assertEquals(200, $response['headers']['status-code']);
+        $this->assertEquals($data['topicId'], $response['body']['topicId']);
+        $this->assertEquals($data['targetId'], $response['body']['targetId']);
+    }
+
+    /**
+     * @depends testCreateSubscriber
+     */
+    public function testListSubscribers(array $data)
+    {
+        $response = $this->client->call(Client::METHOD_GET, '/messaging/topics/' . $data['topicId'] . '/subscribers', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]));
+        $this->assertEquals(200, $response['headers']['status-code']);
+        $this->assertEquals(1, $response['body']['total']);
+        $this->assertEquals(\count($response['body']['subscribers']), $response['body']['total']);
+    }
+
+    /**
+     * @depends testCreateSubscriber
+     */
+    public function testDeleteSubscriber(array $data)
+    {
+        $response = $this->client->call(Client::METHOD_DELETE, '/messaging/topics/' . $data['topicId'] . '/subscriber/' . $data['subscriberId'], \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+        ], $this->getHeaders()));
+        $this->assertEquals(204, $response['headers']['status-code']);
+    }
+
+    /**
+     * @depends testUpdateTopic
+     */
+    public function testDeleteTopic(string $topicId)
+    {
+        $response = $this->client->call(Client::METHOD_DELETE, '/messaging/topics/' . $topicId, [
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]);
+        $this->assertEquals(204, $response['headers']['status-code']);
+    }
+
+    public function testSendEmail()
+    {
+
+        $to = App::getEnv('_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_RECEIVER_EMAIL');
+        $from = App::getEnv('_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_FROM');
+        $apiKey = App::getEnv('_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_API_KEY');
+        $domain = App::getEnv('_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_DOMAIN');
+        $isEuRegion = App::getEnv('_APP_MESSAGE_EMAIL_PROVIDER_MAILGUN_IS_EU_REGION');
+        if (empty($to) || empty($from) || empty($apiKey) || empty($domain) || empty($isEuRegion)) {
+            $this->markTestSkipped('Email provider not configured');
+        }
+
+        // Create provider
+        $provider = $this->client->call(Client::METHOD_POST, '/messaging/providers/mailgun', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]), [
+            'providerId' => ID::unique(),
+            'name' => 'Mailgun-provider',
+            'apiKey' => $apiKey,
+            'domain' => $domain,
+            'isEuRegion' => filter_var($isEuRegion, FILTER_VALIDATE_BOOLEAN),
+            'from' => $from
+        ]);
+        $this->assertEquals(201, $provider['headers']['status-code']);
+
+        // Create Topic
+        $topic = $this->client->call(Client::METHOD_POST, '/messaging/topics', [
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ], [
+            'providerId' => $provider['body']['$id'],
+            'topicId' => ID::unique(),
+            'name' => 'topic1',
+            'description' => 'Test Topic'
+        ]);
+        $this->assertEquals(201, $topic['headers']['status-code']);
+
+        // Create User
+        $user = $this->client->call(Client::METHOD_POST, '/users', [
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ], [
+            'userId' => ID::custom('test-user'),
+            'email' => $to,
+            'password' => 'password',
+            'name' => 'Messaging User',
+        ], false);
+
+        $this->assertEquals(201, $user['headers']['status-code']);
+
+        // Create Target
+        $target = $this->client->call(Client::METHOD_POST, '/users/test-user/targets', [
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ], [
+            'targetId' => ID::unique(),
+            'providerId' => $provider['body']['$id'],
+            'identifier' => $to,
+        ]);
+
+        $this->assertEquals(201, $target['headers']['status-code']);
+
+        // Create Subscriber
+        $subscriber = $this->client->call(Client::METHOD_POST, '/messaging/topics/' . $topic['body']['$id'] . '/subscribers', \array_merge([
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+        ], $this->getHeaders()), [
+            'subscriberId' => ID::unique(),
+            'targetId' => $target['body']['$id'],
+        ]);
+
+        $this->assertEquals(201, $subscriber['headers']['status-code']);
+
+        // Create Email
+        $email = $this->client->call(Client::METHOD_POST, '/messaging/messages/email', [
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ], [
+            'messageId' => ID::unique(),
+            'providerId' => $provider['body']['$id'],
+            'to' => [$topic['body']['$id']],
+            'subject' => 'Khali beats Undertaker',
+            'content' => 'https://www.youtube.com/watch?v=dQw4w9WgXcQ',
+        ]);
+
+        $this->assertEquals(201, $email['headers']['status-code']);
+
+        \sleep(5);
+
+        $message = $this->client->call(Client::METHOD_GET, '/messaging/messages/' . $email['body']['$id'], [
+            'origin' => 'http://localhost',
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ]);
+
+        $this->assertEquals(200, $message['headers']['status-code']);
+        $this->assertEquals(1, $message['body']['deliveredTo']);
+        $this->assertEquals(0, \count($message['body']['deliveryErrors']));
+    }
 }
diff --git a/tests/e2e/Services/Messaging/MessagingConsoleClientTest.php b/tests/e2e/Services/Messaging/MessagingConsoleClientTest.php
new file mode 100644
index 0000000000..0d533c139b
--- /dev/null
+++ b/tests/e2e/Services/Messaging/MessagingConsoleClientTest.php
@@ -0,0 +1,14 @@
+<?php
+
+namespace Tests\E2E\Services\Messaging;
+
+use Tests\E2E\Scopes\ProjectCustom;
+use Tests\E2E\Scopes\Scope;
+use Tests\E2E\Scopes\SideConsole;
+
+class MessagingConsoleClientTest extends Scope
+{
+    use MessagingBase;
+    use ProjectCustom;
+    use SideConsole;
+}
diff --git a/tests/e2e/Services/Messaging/MessagingCustomClientTest.php b/tests/e2e/Services/Messaging/MessagingCustomClientTest.php
new file mode 100644
index 0000000000..ee7d0c2ad9
--- /dev/null
+++ b/tests/e2e/Services/Messaging/MessagingCustomClientTest.php
@@ -0,0 +1,14 @@
+<?php
+
+namespace Tests\E2E\Services\Messaging;
+
+use Tests\E2E\Scopes\ProjectCustom;
+use Tests\E2E\Scopes\Scope;
+use Tests\E2E\Scopes\SideClient;
+
+class MessagingCustomClientTest extends Scope
+{
+    use MessagingBase;
+    use ProjectCustom;
+    use SideClient;
+}