diff --git a/build/rector-strict.php b/build/rector-strict.php
index a4457f40dc22d..a28dab65968a5 100644
--- a/build/rector-strict.php
+++ b/build/rector-strict.php
@@ -6,7 +6,9 @@
  */
 
 use Rector\DeadCode\Rector\ClassMethod\RemoveDuplicatedReturnSelfDocblockRector;
+use Rector\DeadCode\Rector\ClassMethod\RemoveEmptyClassMethodRector;
 use Rector\DeadCode\Rector\ClassMethod\RemoveReturnTagIncompatibleWithNativeTypeRector;
+use Rector\DeadCode\Rector\ClassMethod\RemoveUnusedPublicMethodParameterRector;
 use Rector\Php81\Rector\Property\ReadOnlyPropertyRector;
 use Rector\Php82\Rector\Class_\ReadOnlyClassRector;
 use Rector\PHPUnit\CodeQuality\Rector\Class_\AddSeeTestAnnotationRector;
@@ -63,6 +65,10 @@
 		$nextcloudDir . '/lib/public/SystemReport',
 		$nextcloudDir . '/lib/private/SystemReport',
 		$nextcloudDir . '/tests/lib/SystemReport',
+		$nextcloudDir . '/lib/public/MessageQueue',
+		$nextcloudDir . '/lib/private/MessageQueue',
+		$nextcloudDir . '/core/Command/MessageQueue',
+		$nextcloudDir . '/tests/lib/MessageQueue',
 	])
 	->withAutoloadPaths([
 		// ensure rector properly autoload the public interfaces
@@ -102,4 +108,11 @@
 		// non-final type; removing it breaks psalm's
 		// LessSpecificImplementedReturnType check (psalm-strict).
 		RemoveDuplicatedReturnSelfDocblockRector::class,
+		// Message handler fixtures declare the handled message through their parameter type
+		RemoveUnusedPublicMethodParameterRector::class => [
+			$nextcloudDir . '/tests/lib/MessageQueue/Fixtures',
+		],
+		RemoveEmptyClassMethodRector::class => [
+			$nextcloudDir . '/tests/lib/MessageQueue/Fixtures',
+		],
 	]);
diff --git a/config/config.sample.php b/config/config.sample.php
index c75c85a433718..e9988e3abda4e 100644
--- a/config/config.sample.php
+++ b/config/config.sample.php
@@ -1767,6 +1767,7 @@
 	 * the background jobs which advertise themselves as not time sensitive will be
 	 * delayed during the "working" hours and only run in the 4 hours after the given time.
 	 * This is, e.g., used for activity expiration, suspicious login training, and update checks.
+	 * Messages in the low priority queue of the message queue are delayed the same way.
 	 *
 	 * A value of 1, e.g., will only run these background jobs between 01:00am UTC and 05:00am UTC.
 	 *
@@ -1774,6 +1775,31 @@
 	 */
 	'maintenance_window_start' => 1,
 
+	/**
+	 * Consume the message queue during cron runs
+	 *
+	 * Messages dispatched by apps are handled after the background jobs of each
+	 * cron run. Large instances can disable this and instead run one or more
+	 * dedicated workers with ``occ message-queue:consume``.
+	 *
+	 * Defaults to ``true``
+	 */
+	'message_queue.consume_in_cron' => true,
+
+	/**
+	 * Handle messages right after the request
+	 *
+	 * When running under PHP-FPM, messages dispatched during a web request are
+	 * handled by the same PHP process after the response was sent to the
+	 * client, so they don't have to wait for the next cron run. Messages that
+	 * are not handled within a few seconds stay in the queue. Large instances
+	 * running dedicated workers can disable this to keep PHP-FPM processes
+	 * available for requests.
+	 *
+	 * Defaults to ``true``
+	 */
+	'message_queue.handle_after_request' => true,
+
 	/**
 	 * Log all LDAP requests into a file
 	 *
diff --git a/core/Command/MessageQueue/Consume.php b/core/Command/MessageQueue/Consume.php
new file mode 100644
index 0000000000000..1f6ccc72e56e2
--- /dev/null
+++ b/core/Command/MessageQueue/Consume.php
@@ -0,0 +1,84 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OC\Core\Command\MessageQueue;
+
+use OC\Core\Command\Base;
+use OC\MessageQueue\Consumer;
+use OCP\MessageQueue\Queue;
+use OCP\Util;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface;
+
+final class Consume extends Base {
+	public function __construct(
+		private readonly Consumer $consumer,
+	) {
+		parent::__construct();
+	}
+
+	#[\Override]
+	protected function configure(): void {
+		$this
+			->setName('message-queue:consume')
+			->setDescription('Run a worker consuming messages from the message queue')
+			->addArgument(
+				'queues',
+				InputArgument::OPTIONAL | InputArgument::IS_ARRAY,
+				'Queues to consume, in order of priority (' . implode(', ', array_map(static fn (Queue $queue) => $queue->value, Queue::cases())) . ')',
+			)
+			->addOption('time-limit', 't', InputOption::VALUE_REQUIRED, 'Stop after this many seconds')
+			->addOption('limit', 'l', InputOption::VALUE_REQUIRED, 'Stop after handling this many messages')
+			->addOption('memory-limit', 'm', InputOption::VALUE_REQUIRED, 'Stop when the memory usage exceeds this limit, e.g. 512M')
+			->addOption('sleep', null, InputOption::VALUE_REQUIRED, 'Seconds to wait before polling again when all queues are empty', '1')
+			->addOption('stop-when-empty', null, InputOption::VALUE_NONE, 'Stop as soon as all queues are empty');
+	}
+
+	#[\Override]
+	protected function execute(InputInterface $input, OutputInterface $output): int {
+		/** @var list<string> $names */
+		$names = $input->getArgument('queues');
+		$queues = [];
+		foreach ($names as $name) {
+			$queue = Queue::tryFrom($name);
+			if ($queue === null) {
+				$output->writeln('<error>Unknown queue ' . $name . '</error>');
+				return self::FAILURE;
+			}
+
+			$queues[] = $queue;
+		}
+
+		/** @var mixed $memoryLimit */
+		$memoryLimit = $input->getOption('memory-limit');
+		$this->consumer->consume(
+			queues: $queues ?: Queue::cases(),
+			timeLimit: $this->getIntOption($input, 'time-limit'),
+			messageLimit: $this->getIntOption($input, 'limit'),
+			memoryLimit: is_string($memoryLimit) ? (int)Util::computerFileSize($memoryLimit) : null,
+			stopWhenEmpty: (bool)$input->getOption('stop-when-empty'),
+			sleep: (int)$input->getOption('sleep'),
+			output: $output->isVerbose() ? static fn (string $message) => $output->writeln($message) : null,
+		);
+
+		return self::SUCCESS;
+	}
+
+	#[\Override]
+	public function cancelOperation(): void {
+		parent::cancelOperation();
+		$this->consumer->stop();
+	}
+
+	private function getIntOption(InputInterface $input, string $name): ?int {
+		/** @var mixed $value */
+		$value = $input->getOption($name);
+		return is_string($value) ? (int)$value : null;
+	}
+}
diff --git a/core/Migrations/Version36000Date20261004120000.php b/core/Migrations/Version36000Date20261004120000.php
new file mode 100644
index 0000000000000..6404bc72375cf
--- /dev/null
+++ b/core/Migrations/Version36000Date20261004120000.php
@@ -0,0 +1,62 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+namespace OC\Core\Migrations;
+
+use Closure;
+use OCP\DB\ISchemaWrapper;
+use OCP\DB\Types;
+use OCP\Migration\Attributes\AddIndex;
+use OCP\Migration\Attributes\CreateTable;
+use OCP\Migration\Attributes\IndexType;
+use OCP\Migration\IOutput;
+use OCP\Migration\SimpleMigrationStep;
+use Override;
+
+#[CreateTable(
+	table: 'message_queue',
+	columns: ['id', 'queue_name', 'message_class', 'body', 'retry_count', 'last_error', 'created_at', 'available_at', 'delivered_at', 'deduplication_hash'],
+	description: 'New table to store the messages of the message queue',
+)]
+#[AddIndex(table: 'message_queue', type: IndexType::PRIMARY)]
+#[AddIndex(table: 'message_queue', type: IndexType::INDEX, description: 'Allows to fetch the next available message of a queue')]
+#[AddIndex(table: 'message_queue', type: IndexType::INDEX, description: 'Allows to find pending duplicates of a message')]
+class Version36000Date20261004120000 extends SimpleMigrationStep {
+	#[Override]
+	public function changeSchema(IOutput $output, Closure $schemaClosure, array $options): ?ISchemaWrapper {
+		/** @var ISchemaWrapper $schema */
+		$schema = $schemaClosure();
+
+		if (!$schema->hasTable('message_queue')) {
+			$table = $schema->createTable('message_queue');
+			$table->addColumn('id', Types::BIGINT, [
+				'notnull' => true,
+				'unsigned' => true,
+			]);
+			$table->addColumn('queue_name', Types::STRING, ['notnull' => true, 'length' => 32]);
+			$table->addColumn('message_class', Types::STRING, ['notnull' => true, 'length' => 255]);
+			$table->addColumn('body', Types::TEXT, ['notnull' => true]);
+			$table->addColumn('retry_count', Types::INTEGER, ['notnull' => true, 'default' => 0, 'unsigned' => true]);
+			$table->addColumn('last_error', Types::TEXT, ['notnull' => false]);
+			$table->addColumn('created_at', Types::BIGINT, ['notnull' => true, 'unsigned' => true]);
+			$table->addColumn('available_at', Types::BIGINT, ['notnull' => true, 'unsigned' => true]);
+			$table->addColumn('delivered_at', Types::BIGINT, ['notnull' => false, 'unsigned' => true]);
+			$table->addColumn('deduplication_hash', Types::STRING, ['notnull' => false, 'length' => 64]);
+			$table->setPrimaryKey(['id']);
+			$table->addIndex(['queue_name', 'available_at'], 'mq_queue_available');
+			$table->addIndex(['deduplication_hash'], 'mq_dedup_hash');
+			// Makes sure there is no auto-increment in Oracle
+			$schema->dropAutoincrementColumn('message_queue', 'id');
+
+			return $schema;
+		}
+
+		return null;
+	}
+}
diff --git a/core/Service/CronService.php b/core/Service/CronService.php
index 379b60b10298f..66fb734c54194 100644
--- a/core/Service/CronService.php
+++ b/core/Service/CronService.php
@@ -16,6 +16,7 @@
 use OC\BackgroundJob\JobClassesRegistry;
 use OC\BackgroundJob\JobRuns;
 use OC\DB\Connection;
+use OC\MessageQueue\Consumer;
 use OC\Security\CSRF\TokenStorage\SessionStorage;
 use OC\Session\CryptoWrapper;
 use OC\Session\Memory;
@@ -29,6 +30,7 @@
 use OCP\ILogger;
 use OCP\ISession;
 use OCP\ITempManager;
+use OCP\MessageQueue\Queue;
 use OCP\Util;
 use Psr\Log\LoggerInterface;
 
@@ -54,6 +56,7 @@ public function __construct(
 		private readonly JobRuns $jobRuns,
 		private readonly JobClassesRegistry $jobClassesRegistry,
 		private readonly ISetupManager $setupManager,
+		private readonly Consumer $messageQueueConsumer,
 		private readonly bool $isCLI,
 	) {
 	}
@@ -271,6 +274,11 @@ private function runCli(string $appMode, ?array $jobClasses): void {
 			}
 		}
 
+		if ($jobClasses === null) {
+			$queues = $onlyTimeSensitive ? [Queue::High, Queue::Default] : Queue::cases();
+			$this->consumeMessages($queues, max(60, $endTime - time()));
+		}
+
 		// Makes sure last error isn't caught by shutdown function
 		error_clear_last();
 	}
@@ -287,6 +295,27 @@ private function runWeb(string $appMode): void {
 				$job->start($this->jobList);
 				$this->jobList->setLastJob($job);
 			}
+			$this->consumeMessages(Queue::cases(), 10);
+		}
+	}
+
+	/**
+	 * @param list<Queue> $queues
+	 */
+	private function consumeMessages(array $queues, int $timeLimit): void {
+		if (!$this->config->getSystemValueBool('message_queue.consume_in_cron', true)) {
+			return;
+		}
+
+		try {
+			$this->messageQueueConsumer->consume(
+				queues: $queues,
+				timeLimit: $timeLimit,
+				stopWhenEmpty: true,
+				output: $this->verboseCallback,
+			);
+		} catch (\Throwable $e) {
+			$this->logger->error('Error while consuming the message queue: ' . $e->getMessage(), ['app' => 'cron', 'exception' => $e]);
 		}
 	}
 
diff --git a/core/register_command.php b/core/register_command.php
index d5f8d76279330..27c4acc786f07 100644
--- a/core/register_command.php
+++ b/core/register_command.php
@@ -156,6 +156,7 @@
 	$application->addCommand(Server::get(Delete::class));
 	$application->addCommand(Server::get(JobWorker::class));
 	$application->addCommand(Server::get(RunningJobs::class));
+	$application->addCommand(Server::get(Command\MessageQueue\Consume::class));
 	$application->addCommand(Server::get(JobsHistory::class));
 
 	$application->addCommand(Server::get(Test::class));
diff --git a/lib/composer/composer/autoload_classmap.php b/lib/composer/composer/autoload_classmap.php
index a641ddddb350b..58302ae1e84a8 100644
--- a/lib/composer/composer/autoload_classmap.php
+++ b/lib/composer/composer/autoload_classmap.php
@@ -784,6 +784,13 @@
     'OCP\\Mail\\Provider\\IProvider' => $baseDir . '/lib/public/Mail/Provider/IProvider.php',
     'OCP\\Mail\\Provider\\IService' => $baseDir . '/lib/public/Mail/Provider/IService.php',
     'OCP\\Mail\\Provider\\Message' => $baseDir . '/lib/public/Mail/Provider/Message.php',
+    'OCP\\MessageQueue\\Attribute\\AsMessage' => $baseDir . '/lib/public/MessageQueue/Attribute/AsMessage.php',
+    'OCP\\MessageQueue\\Attribute\\AsMessageHandler' => $baseDir . '/lib/public/MessageQueue/Attribute/AsMessageHandler.php',
+    'OCP\\MessageQueue\\Exception\\InvalidMessageException' => $baseDir . '/lib/public/MessageQueue/Exception/InvalidMessageException.php',
+    'OCP\\MessageQueue\\Exception\\RecoverableMessageException' => $baseDir . '/lib/public/MessageQueue/Exception/RecoverableMessageException.php',
+    'OCP\\MessageQueue\\Exception\\UnrecoverableMessageException' => $baseDir . '/lib/public/MessageQueue/Exception/UnrecoverableMessageException.php',
+    'OCP\\MessageQueue\\IMessageBus' => $baseDir . '/lib/public/MessageQueue/IMessageBus.php',
+    'OCP\\MessageQueue\\Queue' => $baseDir . '/lib/public/MessageQueue/Queue.php',
     'OCP\\Migration\\Attributes\\AddColumn' => $baseDir . '/lib/public/Migration/Attributes/AddColumn.php',
     'OCP\\Migration\\Attributes\\AddIndex' => $baseDir . '/lib/public/Migration/Attributes/AddIndex.php',
     'OCP\\Migration\\Attributes\\ColumnMigrationAttribute' => $baseDir . '/lib/public/Migration/Attributes/ColumnMigrationAttribute.php',
@@ -1534,6 +1541,7 @@
     'OC\\Core\\Command\\Memcache\\DistributedGet' => $baseDir . '/core/Command/Memcache/DistributedGet.php',
     'OC\\Core\\Command\\Memcache\\DistributedSet' => $baseDir . '/core/Command/Memcache/DistributedSet.php',
     'OC\\Core\\Command\\Memcache\\RedisCommand' => $baseDir . '/core/Command/Memcache/RedisCommand.php',
+    'OC\\Core\\Command\\MessageQueue\\Consume' => $baseDir . '/core/Command/MessageQueue/Consume.php',
     'OC\\Core\\Command\\OCM\\ActivateKey' => $baseDir . '/core/Command/OCM/ActivateKey.php',
     'OC\\Core\\Command\\OCM\\ListKeys' => $baseDir . '/core/Command/OCM/ListKeys.php',
     'OC\\Core\\Command\\OCM\\RetireKey' => $baseDir . '/core/Command/OCM/RetireKey.php',
@@ -1743,6 +1751,7 @@
     'OC\\Core\\Migrations\\Version34000Date20260518163022' => $baseDir . '/core/Migrations/Version34000Date20260518163022.php',
     'OC\\Core\\Migrations\\Version34000Date20260521110333' => $baseDir . '/core/Migrations/Version34000Date20260521110333.php',
     'OC\\Core\\Migrations\\Version35000Date20260527162338' => $baseDir . '/core/Migrations/Version35000Date20260527162338.php',
+    'OC\\Core\\Migrations\\Version36000Date20261004120000' => $baseDir . '/core/Migrations/Version36000Date20261004120000.php',
     'OC\\Core\\Notification\\CoreNotifier' => $baseDir . '/core/Notification/CoreNotifier.php',
     'OC\\Core\\ResponseDefinitions' => $baseDir . '/core/ResponseDefinitions.php',
     'OC\\Core\\Service\\CronService' => $baseDir . '/core/Service/CronService.php',
@@ -2098,6 +2107,14 @@
     'OC\\Memcache\\Redis' => $baseDir . '/lib/private/Memcache/Redis.php',
     'OC\\Memcache\\WithLocalCache' => $baseDir . '/lib/private/Memcache/WithLocalCache.php',
     'OC\\MemoryInfo' => $baseDir . '/lib/private/MemoryInfo.php',
+    'OC\\MessageQueue\\AfterRequestHandler' => $baseDir . '/lib/private/MessageQueue/AfterRequestHandler.php',
+    'OC\\MessageQueue\\Consumer' => $baseDir . '/lib/private/MessageQueue/Consumer.php',
+    'OC\\MessageQueue\\HandlerRegistry' => $baseDir . '/lib/private/MessageQueue/HandlerRegistry.php',
+    'OC\\MessageQueue\\MessageBus' => $baseDir . '/lib/private/MessageQueue/MessageBus.php',
+    'OC\\MessageQueue\\MessageMetadataReader' => $baseDir . '/lib/private/MessageQueue/MessageMetadataReader.php',
+    'OC\\MessageQueue\\MessageNormalizer' => $baseDir . '/lib/private/MessageQueue/MessageNormalizer.php',
+    'OC\\MessageQueue\\MessageStore' => $baseDir . '/lib/private/MessageQueue/MessageStore.php',
+    'OC\\MessageQueue\\QueuedMessage' => $baseDir . '/lib/private/MessageQueue/QueuedMessage.php',
     'OC\\Migration\\BackgroundRepair' => $baseDir . '/lib/private/Migration/BackgroundRepair.php',
     'OC\\Migration\\ConsoleOutput' => $baseDir . '/lib/private/Migration/ConsoleOutput.php',
     'OC\\Migration\\Exceptions\\AttributeException' => $baseDir . '/lib/private/Migration/Exceptions/AttributeException.php',
diff --git a/lib/composer/composer/autoload_static.php b/lib/composer/composer/autoload_static.php
index 1931ad3bff8a3..f0824192967bc 100644
--- a/lib/composer/composer/autoload_static.php
+++ b/lib/composer/composer/autoload_static.php
@@ -830,6 +830,13 @@ class ComposerStaticInit749170dad3f5e7f9ca158f5a9f04f6a2
         'OCP\\Mail\\Provider\\IProvider' => __DIR__ . '/../../..' . '/lib/public/Mail/Provider/IProvider.php',
         'OCP\\Mail\\Provider\\IService' => __DIR__ . '/../../..' . '/lib/public/Mail/Provider/IService.php',
         'OCP\\Mail\\Provider\\Message' => __DIR__ . '/../../..' . '/lib/public/Mail/Provider/Message.php',
+        'OCP\\MessageQueue\\Attribute\\AsMessage' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/Attribute/AsMessage.php',
+        'OCP\\MessageQueue\\Attribute\\AsMessageHandler' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/Attribute/AsMessageHandler.php',
+        'OCP\\MessageQueue\\Exception\\InvalidMessageException' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/Exception/InvalidMessageException.php',
+        'OCP\\MessageQueue\\Exception\\RecoverableMessageException' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/Exception/RecoverableMessageException.php',
+        'OCP\\MessageQueue\\Exception\\UnrecoverableMessageException' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/Exception/UnrecoverableMessageException.php',
+        'OCP\\MessageQueue\\IMessageBus' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/IMessageBus.php',
+        'OCP\\MessageQueue\\Queue' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/Queue.php',
         'OCP\\Migration\\Attributes\\AddColumn' => __DIR__ . '/../../..' . '/lib/public/Migration/Attributes/AddColumn.php',
         'OCP\\Migration\\Attributes\\AddIndex' => __DIR__ . '/../../..' . '/lib/public/Migration/Attributes/AddIndex.php',
         'OCP\\Migration\\Attributes\\ColumnMigrationAttribute' => __DIR__ . '/../../..' . '/lib/public/Migration/Attributes/ColumnMigrationAttribute.php',
@@ -1580,6 +1587,7 @@ class ComposerStaticInit749170dad3f5e7f9ca158f5a9f04f6a2
         'OC\\Core\\Command\\Memcache\\DistributedGet' => __DIR__ . '/../../..' . '/core/Command/Memcache/DistributedGet.php',
         'OC\\Core\\Command\\Memcache\\DistributedSet' => __DIR__ . '/../../..' . '/core/Command/Memcache/DistributedSet.php',
         'OC\\Core\\Command\\Memcache\\RedisCommand' => __DIR__ . '/../../..' . '/core/Command/Memcache/RedisCommand.php',
+        'OC\\Core\\Command\\MessageQueue\\Consume' => __DIR__ . '/../../..' . '/core/Command/MessageQueue/Consume.php',
         'OC\\Core\\Command\\OCM\\ActivateKey' => __DIR__ . '/../../..' . '/core/Command/OCM/ActivateKey.php',
         'OC\\Core\\Command\\OCM\\ListKeys' => __DIR__ . '/../../..' . '/core/Command/OCM/ListKeys.php',
         'OC\\Core\\Command\\OCM\\RetireKey' => __DIR__ . '/../../..' . '/core/Command/OCM/RetireKey.php',
@@ -1789,6 +1797,7 @@ class ComposerStaticInit749170dad3f5e7f9ca158f5a9f04f6a2
         'OC\\Core\\Migrations\\Version34000Date20260518163022' => __DIR__ . '/../../..' . '/core/Migrations/Version34000Date20260518163022.php',
         'OC\\Core\\Migrations\\Version34000Date20260521110333' => __DIR__ . '/../../..' . '/core/Migrations/Version34000Date20260521110333.php',
         'OC\\Core\\Migrations\\Version35000Date20260527162338' => __DIR__ . '/../../..' . '/core/Migrations/Version35000Date20260527162338.php',
+        'OC\\Core\\Migrations\\Version36000Date20261004120000' => __DIR__ . '/../../..' . '/core/Migrations/Version36000Date20261004120000.php',
         'OC\\Core\\Notification\\CoreNotifier' => __DIR__ . '/../../..' . '/core/Notification/CoreNotifier.php',
         'OC\\Core\\ResponseDefinitions' => __DIR__ . '/../../..' . '/core/ResponseDefinitions.php',
         'OC\\Core\\Service\\CronService' => __DIR__ . '/../../..' . '/core/Service/CronService.php',
@@ -2144,6 +2153,14 @@ class ComposerStaticInit749170dad3f5e7f9ca158f5a9f04f6a2
         'OC\\Memcache\\Redis' => __DIR__ . '/../../..' . '/lib/private/Memcache/Redis.php',
         'OC\\Memcache\\WithLocalCache' => __DIR__ . '/../../..' . '/lib/private/Memcache/WithLocalCache.php',
         'OC\\MemoryInfo' => __DIR__ . '/../../..' . '/lib/private/MemoryInfo.php',
+        'OC\\MessageQueue\\AfterRequestHandler' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/AfterRequestHandler.php',
+        'OC\\MessageQueue\\Consumer' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/Consumer.php',
+        'OC\\MessageQueue\\HandlerRegistry' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/HandlerRegistry.php',
+        'OC\\MessageQueue\\MessageBus' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/MessageBus.php',
+        'OC\\MessageQueue\\MessageMetadataReader' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/MessageMetadataReader.php',
+        'OC\\MessageQueue\\MessageNormalizer' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/MessageNormalizer.php',
+        'OC\\MessageQueue\\MessageStore' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/MessageStore.php',
+        'OC\\MessageQueue\\QueuedMessage' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/QueuedMessage.php',
         'OC\\Migration\\BackgroundRepair' => __DIR__ . '/../../..' . '/lib/private/Migration/BackgroundRepair.php',
         'OC\\Migration\\ConsoleOutput' => __DIR__ . '/../../..' . '/lib/private/Migration/ConsoleOutput.php',
         'OC\\Migration\\Exceptions\\AttributeException' => __DIR__ . '/../../..' . '/lib/private/Migration/Exceptions/AttributeException.php',
diff --git a/lib/private/App/InfoParser.php b/lib/private/App/InfoParser.php
index 94e8951fd372e..66b12c2685482 100644
--- a/lib/private/App/InfoParser.php
+++ b/lib/private/App/InfoParser.php
@@ -93,6 +93,9 @@ public function parse(string $file): ?array {
 		if (!array_key_exists('commands', $array)) {
 			$array['commands'] = [];
 		}
+		if (!array_key_exists('message-handlers', $array)) {
+			$array['message-handlers'] = [];
+		}
 		if (!array_key_exists('activity', $array)) {
 			$array['activity'] = [];
 		}
@@ -160,6 +163,9 @@ public function parse(string $file): ?array {
 		if (isset($array['commands']['command']) && is_array($array['commands']['command'])) {
 			$array['commands'] = $array['commands']['command'];
 		}
+		if (isset($array['message-handlers']['handler']) && is_array($array['message-handlers']['handler'])) {
+			$array['message-handlers'] = $array['message-handlers']['handler'];
+		}
 		if (isset($array['two-factor-providers']['provider']) && is_array($array['two-factor-providers']['provider'])) {
 			$array['two-factor-providers'] = $array['two-factor-providers']['provider'];
 		}
diff --git a/lib/private/MessageQueue/AfterRequestHandler.php b/lib/private/MessageQueue/AfterRequestHandler.php
new file mode 100644
index 0000000000000..eb3e4204c8441
--- /dev/null
+++ b/lib/private/MessageQueue/AfterRequestHandler.php
@@ -0,0 +1,96 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OC\MessageQueue;
+
+use OCP\Files\ISetupManager;
+use OCP\IConfig;
+use OCP\ISession;
+use OCP\IUserSession;
+use OCP\Lock\ILockingProvider;
+use OCP\MessageQueue\Queue;
+use Psr\Log\LoggerInterface;
+
+/**
+ * Handles the messages dispatched during a web request once the response has
+ * been sent to the client, using fastcgi_finish_request(). Messages that are
+ * not handled within the budget stay in the queue for cron and workers.
+ */
+class AfterRequestHandler {
+	private const int TIME_LIMIT = 10;
+
+	private const int MESSAGE_LIMIT = 5;
+
+	/** @var list<string> */
+	private array $messageIds = [];
+
+	private bool $registered = false;
+
+	public function __construct(
+		private readonly Consumer $consumer,
+		private readonly IConfig $config,
+		private readonly ISession $session,
+		private readonly IUserSession $userSession,
+		private readonly ISetupManager $setupManager,
+		private readonly ILockingProvider $lockingProvider,
+		private readonly LoggerInterface $logger,
+		private readonly bool $isCLI,
+	) {
+	}
+
+	public function add(Queue $queue, string $messageId): void {
+		if ($queue === Queue::Low || !$this->isEnabled()) {
+			return;
+		}
+
+		$this->messageIds[] = $messageId;
+		if (!$this->registered) {
+			$this->registered = true;
+			$this->registerShutdownFunction($this->run(...));
+		}
+	}
+
+	public function run(): void {
+		$messageIds = $this->messageIds;
+		$this->messageIds = [];
+		if ($messageIds === []) {
+			return;
+		}
+
+		try {
+			$this->session->close();
+			$this->finishRequest();
+
+			$this->userSession->setVolatileActiveUser(null);
+			$this->setupManager->tearDown();
+			$this->consumer->consumeMessages($messageIds, self::TIME_LIMIT, self::MESSAGE_LIMIT);
+		} catch (\Throwable $throwable) {
+			$this->logger->error('Error while handling messages after the request: ' . $throwable->getMessage(), ['exception' => $throwable]);
+		} finally {
+			// The shutdown functions registered during boot already ran
+			$this->lockingProvider->releaseAll();
+		}
+	}
+
+	protected function isEnabled(): bool {
+		return !$this->isCLI
+			&& $this->canFinishRequest()
+			&& $this->config->getSystemValueBool('message_queue.handle_after_request', true);
+	}
+
+	protected function canFinishRequest(): bool {
+		return function_exists('fastcgi_finish_request');
+	}
+
+	protected function registerShutdownFunction(callable $callback): void {
+		register_shutdown_function($callback);
+	}
+
+	protected function finishRequest(): void {
+		fastcgi_finish_request();
+	}
+}
diff --git a/lib/private/MessageQueue/Consumer.php b/lib/private/MessageQueue/Consumer.php
new file mode 100644
index 0000000000000..01fba64445fd6
--- /dev/null
+++ b/lib/private/MessageQueue/Consumer.php
@@ -0,0 +1,269 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OC\MessageQueue;
+
+use OCP\Files\ISetupManager;
+use OCP\IDBConnection;
+use OCP\ITempManager;
+use OCP\IUser;
+use OCP\IUserManager;
+use OCP\IUserSession;
+use OCP\MessageQueue\Exception\InvalidMessageException;
+use OCP\MessageQueue\Exception\RecoverableMessageException;
+use OCP\MessageQueue\Exception\UnrecoverableMessageException;
+use OCP\MessageQueue\Queue;
+use Psr\Container\ContainerInterface;
+use Psr\Log\LoggerInterface;
+
+/**
+ * Consumes messages from the given queues, highest priority first.
+ *
+ * @psalm-suppress ClassMustBeFinal For unit tests
+ */
+class Consumer {
+	private const int MAX_RETRY_DELAY = 86400;
+
+	private bool $shouldStop = false;
+
+	public function __construct(
+		private readonly MessageStore $store,
+		private readonly HandlerRegistry $handlerRegistry,
+		private readonly MessageMetadataReader $metadataReader,
+		private readonly MessageNormalizer $normalizer,
+		private readonly ContainerInterface $container,
+		private readonly IUserManager $userManager,
+		private readonly IUserSession $userSession,
+		private readonly ISetupManager $setupManager,
+		private readonly IDBConnection $connection,
+		private readonly ITempManager $tempManager,
+		private readonly LoggerInterface $logger,
+	) {
+	}
+
+	/**
+	 * @param list<Queue> $queues
+	 * @param int|null $timeLimit Stop after this many seconds
+	 * @param int|null $messageLimit Stop after handling this many messages
+	 * @param int|null $memoryLimit Stop when the memory usage exceeds this many bytes
+	 * @param bool $stopWhenEmpty Stop as soon as all queues are empty instead of waiting for new messages
+	 * @param int $sleep Seconds to wait before polling again when all queues are empty
+	 * @param (callable(string): void)|null $output Receives progress messages
+	 */
+	public function consume(
+		array $queues = [Queue::High, Queue::Default, Queue::Low],
+		?int $timeLimit = null,
+		?int $messageLimit = null,
+		?int $memoryLimit = null,
+		bool $stopWhenEmpty = false,
+		int $sleep = 1,
+		?callable $output = null,
+	): void {
+		$next = function () use ($queues): ?QueuedMessage {
+			foreach ($queues as $queue) {
+				$message = $this->store->claimNext($queue);
+				if ($message instanceof QueuedMessage) {
+					return $message;
+				}
+			}
+
+			return null;
+		};
+		$this->run($next, $timeLimit, $messageLimit, $memoryLimit, $stopWhenEmpty, $sleep, $output);
+	}
+
+	/**
+	 * Handles the given messages in order, if they are still available.
+	 *
+	 * @param list<string> $messageIds
+	 */
+	public function consumeMessages(array $messageIds, ?int $timeLimit = null, ?int $messageLimit = null): void {
+		$next = function () use (&$messageIds): ?QueuedMessage {
+			while (($id = array_shift($messageIds)) !== null) {
+				$message = $this->store->claimById($id);
+				if ($message instanceof QueuedMessage) {
+					return $message;
+				}
+			}
+
+			return null;
+		};
+		$this->run($next, $timeLimit, $messageLimit, stopWhenEmpty: true);
+	}
+
+	/**
+	 * Stops consuming after the current message.
+	 */
+	public function stop(): void {
+		$this->shouldStop = true;
+	}
+
+	/**
+	 * @param \Closure(): ?QueuedMessage $next
+	 * @param (callable(string): void)|null $output
+	 */
+	private function run(
+		\Closure $next,
+		?int $timeLimit = null,
+		?int $messageLimit = null,
+		?int $memoryLimit = null,
+		bool $stopWhenEmpty = false,
+		int $sleep = 1,
+		?callable $output = null,
+	): void {
+		$endTime = $timeLimit !== null ? microtime(true) + (float)$timeLimit : null;
+		$handled = 0;
+
+		try {
+			while (!$this->shouldStop) {
+				if ($endTime !== null && microtime(true) >= $endTime) {
+					break;
+				}
+
+				$message = $next();
+				if ($message instanceof QueuedMessage) {
+					$this->handle($message, $output);
+					++$handled;
+					if ($messageLimit !== null && $handled >= $messageLimit) {
+						break;
+					}
+
+					if ($memoryLimit !== null && memory_get_usage(true) > $memoryLimit) {
+						break;
+					}
+				} elseif ($stopWhenEmpty) {
+					break;
+				} else {
+					sleep($sleep);
+				}
+
+				$this->dispatchSignals();
+			}
+		} finally {
+			$this->shouldStop = false;
+		}
+	}
+
+	/**
+	 * Runs the signal handlers, which may call stop().
+	 */
+	private function dispatchSignals(): void {
+		if (function_exists('pcntl_signal_dispatch')) {
+			pcntl_signal_dispatch();
+		}
+	}
+
+	/**
+	 * @param (callable(string): void)|null $output
+	 */
+	private function handle(QueuedMessage $queued, ?callable $output): void {
+		if ($output !== null) {
+			$output('Handling ' . $queued->messageClass . ' (' . $queued->id . ')');
+		}
+
+		try {
+			$this->invokeHandler($queued);
+			$error = null;
+		} catch (\Throwable $throwable) {
+			$error = $throwable;
+		}
+
+		// Before writing to the store, so a transaction left open by the handler can't roll it back
+		$this->cleanUp();
+
+		if (!$error instanceof \Throwable) {
+			$this->store->delete($queued->id);
+			return;
+		}
+
+		$delay = $this->getRetryDelay($queued, $error);
+		if ($delay !== null) {
+			$this->logger->warning('Error while handling message ' . $queued->messageClass . ', retrying in ' . $delay . ' seconds: ' . $error->getMessage(), ['exception' => $error]);
+			$this->store->retry($queued->id, $delay);
+		} else {
+			$this->logger->error('Error while handling message ' . $queued->messageClass . ', moving it to the failed queue: ' . $error->getMessage(), ['exception' => $error]);
+			$this->store->fail($queued->id, $error::class . ': ' . $error->getMessage());
+		}
+
+		if ($output !== null) {
+			$output('Failed ' . $queued->messageClass . ': ' . $error->getMessage() . ($delay !== null ? ' (will retry)' : ''));
+		}
+	}
+
+	private function invokeHandler(QueuedMessage $queued): void {
+		$class = $queued->messageClass;
+		$handler = $this->handlerRegistry->getHandler($class);
+		if ($handler === null || !$this->metadataReader->isMessage($class)) {
+			throw new UnrecoverableMessageException('No handler registered for message ' . $class);
+		}
+
+		try {
+			$data = json_decode($queued->body, true, flags: JSON_THROW_ON_ERROR);
+			if (!is_array($data)) {
+				throw new InvalidMessageException('Message body is not an object');
+			}
+
+			$message = $this->normalizer->denormalize($class, $data);
+		} catch (\JsonException|InvalidMessageException $e) {
+			throw new UnrecoverableMessageException('Could not decode message ' . $class . ': ' . $e->getMessage(), $e->getCode(), previous: $e);
+		}
+
+		$userProperty = $this->metadataReader->get($class)->userProperty;
+		$user = null;
+		if ($userProperty !== null) {
+			/** @var mixed $userId */
+			$userId = $message->$userProperty;
+			$user = is_string($userId) ? $this->userManager->get($userId) : null;
+			if (!$user instanceof IUser) {
+				throw new UnrecoverableMessageException('User ' . var_export($userId, true) . ' of message ' . $class . ' does not exist');
+			}
+
+		}
+
+		try {
+			if ($user instanceof IUser) {
+				$this->userSession->setVolatileActiveUser($user);
+				$this->setupManager->setupForUser($user);
+			}
+
+			$callable = [$this->container->get($handler['class']), $handler['method']];
+			if (!is_callable($callable)) {
+				throw new UnrecoverableMessageException('Message handler ' . $handler['class'] . '::' . $handler['method'] . ' is not callable');
+			}
+
+			$callable($message);
+		} finally {
+			$this->setupManager->tearDown();
+			$this->userSession->setVolatileActiveUser(null);
+		}
+	}
+
+	/**
+	 * @return int|null Seconds to wait before retrying, null if the message should not be retried
+	 */
+	private function getRetryDelay(QueuedMessage $queued, \Throwable $error): ?int {
+		if ($error instanceof UnrecoverableMessageException) {
+			return null;
+		}
+
+		$metadata = $this->metadataReader->get($queued->messageClass);
+		if (!$error instanceof RecoverableMessageException && $queued->retryCount >= $metadata->maxRetries) {
+			return null;
+		}
+
+		$delay = (float)$metadata->retryDelay * ($metadata->retryMultiplier ** (float)$queued->retryCount);
+		return (int)min((float)self::MAX_RETRY_DELAY, $delay);
+	}
+
+	private function cleanUp(): void {
+		$this->tempManager->clean();
+		if ($this->connection->inTransaction()) {
+			$this->connection->rollBack();
+			$this->logger->warning('A message handler left a transaction open, it was rolled back');
+		}
+	}
+}
diff --git a/lib/private/MessageQueue/HandlerRegistry.php b/lib/private/MessageQueue/HandlerRegistry.php
new file mode 100644
index 0000000000000..4011603229664
--- /dev/null
+++ b/lib/private/MessageQueue/HandlerRegistry.php
@@ -0,0 +1,125 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OC\MessageQueue;
+
+use OCP\App\IAppManager;
+use OCP\ICache;
+use OCP\ICacheFactory;
+use OCP\MessageQueue\Attribute\AsMessageHandler;
+use Psr\Log\LoggerInterface;
+
+/**
+ * Maps message classes to their handler, as declared with #[AsMessageHandler]
+ * on the classes listed in the <message-handlers> section of enabled apps.
+ *
+ * @psalm-type Handler = array{class: class-string, method: string}
+ * @psalm-suppress ClassMustBeFinal For unit tests
+ */
+class HandlerRegistry {
+	/** @var array<string, Handler>|null */
+	private ?array $handlers = null;
+
+	private readonly ICache $cache;
+
+	public function __construct(
+		private readonly IAppManager $appManager,
+		private readonly MessageMetadataReader $metadataReader,
+		private readonly LoggerInterface $logger,
+		ICacheFactory $cacheFactory,
+	) {
+		$this->cache = $cacheFactory->createLocal('message-queue');
+	}
+
+	/**
+	 * @return Handler|null
+	 */
+	public function getHandler(string $messageClass): ?array {
+		return $this->getHandlers()[$messageClass] ?? null;
+	}
+
+	/**
+	 * @return array<string, Handler>
+	 */
+	public function getHandlers(): array {
+		if ($this->handlers !== null) {
+			return $this->handlers;
+		}
+
+		$apps = [];
+		foreach ($this->appManager->getEnabledApps() as $app) {
+			$apps[$app] = $this->appManager->getAppVersion($app);
+		}
+
+		$cacheKey = 'handlers-' . md5(json_encode($apps, JSON_THROW_ON_ERROR));
+
+		/** @var array<string, Handler>|null $cached */
+		$cached = $this->cache->get($cacheKey);
+		if (is_array($cached)) {
+			return $this->handlers = $cached;
+		}
+
+		$handlers = [];
+		foreach (array_keys($apps) as $app) {
+			$info = $this->appManager->getAppInfo($app);
+			foreach ($info['message-handlers'] ?? [] as $handlerClass) {
+				$this->collect($app, $handlerClass, $handlers);
+			}
+		}
+
+		$this->cache->set($cacheKey, $handlers, 3600);
+		return $this->handlers = $handlers;
+	}
+
+	/**
+	 * @param array<string, Handler> $handlers
+	 */
+	private function collect(string $app, string $handlerClass, array &$handlers): void {
+		if (!class_exists($handlerClass)) {
+			$this->logger->error('Message handler ' . $handlerClass . ' of app ' . $app . ' does not exist');
+			return;
+		}
+
+		$class = new \ReflectionClass($handlerClass);
+		$methods = [];
+		if ($class->getAttributes(AsMessageHandler::class) !== []) {
+			if (!$class->hasMethod('__invoke')) {
+				$this->logger->error('Message handler ' . $handlerClass . ' of app ' . $app . ' is marked with #[AsMessageHandler] but has no __invoke method');
+				return;
+			}
+
+			$methods['__invoke'] = $class->getMethod('__invoke');
+		}
+
+		foreach ($class->getMethods(\ReflectionMethod::IS_PUBLIC) as $method) {
+			if ($method->getAttributes(AsMessageHandler::class) !== []) {
+				$methods[$method->getName()] = $method;
+			}
+		}
+
+		if ($methods === []) {
+			$this->logger->error('Message handler ' . $handlerClass . ' of app ' . $app . ' has no #[AsMessageHandler] attribute');
+			return;
+		}
+
+		foreach ($methods as $method) {
+			$type = ($method->getParameters()[0] ?? null)?->getType();
+			$messageClass = $type instanceof \ReflectionNamedType ? $type->getName() : null;
+			if ($messageClass === null || !$this->metadataReader->isMessage($messageClass)) {
+				$this->logger->error('First parameter of message handler ' . $handlerClass . '::' . $method->getName() . ' must be a class marked with #[AsMessage]');
+				continue;
+			}
+
+			if (isset($handlers[$messageClass])) {
+				$this->logger->error('Message ' . $messageClass . ' already has a handler, ignoring ' . $handlerClass . '::' . $method->getName());
+				continue;
+			}
+
+			$handlers[$messageClass] = ['class' => $handlerClass, 'method' => $method->getName()];
+		}
+	}
+}
diff --git a/lib/private/MessageQueue/MessageBus.php b/lib/private/MessageQueue/MessageBus.php
new file mode 100644
index 0000000000000..1ecdd7882d23d
--- /dev/null
+++ b/lib/private/MessageQueue/MessageBus.php
@@ -0,0 +1,45 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OC\MessageQueue;
+
+use OCP\MessageQueue\Exception\InvalidMessageException;
+use OCP\MessageQueue\IMessageBus;
+
+final readonly class MessageBus implements IMessageBus {
+	public function __construct(
+		private MessageStore $store,
+		private HandlerRegistry $handlerRegistry,
+		private MessageMetadataReader $metadataReader,
+		private MessageNormalizer $normalizer,
+		private AfterRequestHandler $afterRequestHandler,
+	) {
+	}
+
+	#[\Override]
+	public function dispatch(object $message): void {
+		$class = $message::class;
+		if ($this->handlerRegistry->getHandler($class) === null) {
+			throw new InvalidMessageException('No handler registered for message ' . $class);
+		}
+
+		$metadata = $this->metadataReader->get($class);
+		$data = $this->normalizer->normalize($message);
+
+		$deduplicationHash = null;
+		if ($metadata->deduplicateBy !== []) {
+			$key = array_intersect_key($data, array_flip($metadata->deduplicateBy));
+			ksort($key);
+			$deduplicationHash = hash('sha256', $class . json_encode($key, JSON_THROW_ON_ERROR));
+		}
+
+		$id = $this->store->add($metadata->queue, $class, json_encode($data, JSON_THROW_ON_ERROR), $deduplicationHash);
+		if ($id !== null) {
+			$this->afterRequestHandler->add($metadata->queue, $id);
+		}
+	}
+}
diff --git a/lib/private/MessageQueue/MessageMetadataReader.php b/lib/private/MessageQueue/MessageMetadataReader.php
new file mode 100644
index 0000000000000..fd229bd4d33a6
--- /dev/null
+++ b/lib/private/MessageQueue/MessageMetadataReader.php
@@ -0,0 +1,48 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OC\MessageQueue;
+
+use OCP\MessageQueue\Attribute\AsMessage;
+use OCP\MessageQueue\Exception\InvalidMessageException;
+
+final class MessageMetadataReader {
+	/** @var array<string, AsMessage|null> */
+	private array $cache = [];
+
+	/**
+	 * @psalm-assert-if-true class-string $class
+	 */
+	public function isMessage(string $class): bool {
+		return $this->read($class) instanceof AsMessage;
+	}
+
+	/**
+	 * @throws InvalidMessageException
+	 */
+	public function get(string $class): AsMessage {
+		$metadata = $this->read($class);
+		if (!$metadata instanceof AsMessage) {
+			throw new InvalidMessageException('Class ' . $class . ' is not marked with #[AsMessage]');
+		}
+
+		return $metadata;
+	}
+
+	private function read(string $class): ?AsMessage {
+		if (!array_key_exists($class, $this->cache)) {
+			if (!class_exists($class)) {
+				return null;
+			}
+
+			$attributes = (new \ReflectionClass($class))->getAttributes(AsMessage::class);
+			$this->cache[$class] = $attributes === [] ? null : $attributes[0]->newInstance();
+		}
+
+		return $this->cache[$class];
+	}
+}
diff --git a/lib/private/MessageQueue/MessageNormalizer.php b/lib/private/MessageQueue/MessageNormalizer.php
new file mode 100644
index 0000000000000..e7415c0082c89
--- /dev/null
+++ b/lib/private/MessageQueue/MessageNormalizer.php
@@ -0,0 +1,142 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OC\MessageQueue;
+
+use OCP\MessageQueue\Exception\InvalidMessageException;
+
+/**
+ * Converts messages from and to arrays of plain values, based on their
+ * constructor signature. Class names are never read from the payload.
+ */
+final readonly class MessageNormalizer {
+	private const string DATE_FORMAT = 'Y-m-d\TH:i:s.uP';
+
+	public function __construct(
+		private MessageMetadataReader $metadataReader,
+	) {
+	}
+
+	/**
+	 * @return array<string, mixed>
+	 * @throws InvalidMessageException
+	 */
+	public function normalize(object $message): array {
+		$class = $message::class;
+		$data = [];
+		foreach ($this->getConstructorParameters($class) as $parameter) {
+			$name = $parameter->getName();
+			if (!property_exists($message, $name) || !(new \ReflectionProperty($message, $name))->isPublic()) {
+				throw new InvalidMessageException('Constructor parameter $' . $name . ' of ' . $class . ' must be a public promoted property');
+			}
+
+			$data[$name] = $this->normalizeValue($message->$name, $class . '::$' . $name);
+		}
+
+		return $data;
+	}
+
+	/**
+	 * @template T of object
+	 * @param class-string<T> $class
+	 * @param array<array-key, mixed> $data
+	 * @return T
+	 * @throws InvalidMessageException
+	 */
+	public function denormalize(string $class, array $data): object {
+		$arguments = [];
+		foreach ($this->getConstructorParameters($class) as $parameter) {
+			$name = $parameter->getName();
+			if (!array_key_exists($name, $data)) {
+				if ($parameter->isDefaultValueAvailable()) {
+					continue;
+				}
+
+				throw new InvalidMessageException('Missing value for ' . $class . '::$' . $name);
+			}
+
+			$arguments[$name] = $this->denormalizeValue($data[$name], $parameter->getType(), $class . '::$' . $name);
+		}
+
+		return (new \ReflectionClass($class))->newInstanceArgs($arguments);
+	}
+
+	/**
+	 * @param class-string $class
+	 * @return list<\ReflectionParameter>
+	 */
+	private function getConstructorParameters(string $class): array {
+		return (new \ReflectionClass($class))->getConstructor()?->getParameters() ?? [];
+	}
+
+	private function normalizeValue(mixed $value, string $path, bool $inArray = false): int|float|string|bool|array|null {
+		if ($value === null || is_scalar($value)) {
+			return $value;
+		}
+
+		if (is_array($value)) {
+			return array_map(fn (mixed $item): int|float|string|bool|array|null => $this->normalizeValue($item, $path, true), $value);
+		}
+
+		if (!$inArray && is_object($value)) {
+			if ($value instanceof \BackedEnum) {
+				return $value->value;
+			}
+
+			if ($value instanceof \DateTimeImmutable) {
+				return $value->format(self::DATE_FORMAT);
+			}
+
+			if ($this->metadataReader->isMessage($value::class)) {
+				return $this->normalize($value);
+			}
+		}
+
+		throw new InvalidMessageException('Unsupported value of type ' . get_debug_type($value) . ' in ' . $path);
+	}
+
+	private function denormalizeValue(mixed $value, ?\ReflectionType $type, string $path): int|float|string|bool|array|object|null {
+		if (!$type instanceof \ReflectionNamedType) {
+			throw new InvalidMessageException('A single named type is required in ' . $path);
+		}
+
+		if ($value === null) {
+			if ($type->allowsNull()) {
+				return null;
+			}
+
+			throw new InvalidMessageException('Unexpected null value in ' . $path);
+		}
+
+		$typeName = $type->getName();
+		if ($type->isBuiltin()) {
+			return match (true) {
+				$typeName === 'int' && is_int($value) => $value,
+				$typeName === 'float' && (is_float($value) || is_int($value)) => (float)$value,
+				$typeName === 'string' && is_string($value) => $value,
+				$typeName === 'bool' && is_bool($value) => $value,
+				$typeName === 'array' && is_array($value) => $value,
+				default => throw new InvalidMessageException('Expected ' . $typeName . ' but got ' . get_debug_type($value) . ' in ' . $path),
+			};
+		}
+
+		if (is_subclass_of($typeName, \BackedEnum::class) && (is_int($value) || is_string($value))) {
+			return $typeName::tryFrom($value) ?? throw new InvalidMessageException('Invalid enum value in ' . $path);
+		}
+
+		if (in_array($typeName, [\DateTimeImmutable::class, \DateTimeInterface::class], true) && is_string($value)) {
+			return \DateTimeImmutable::createFromFormat(self::DATE_FORMAT, $value)
+				?: throw new InvalidMessageException('Invalid date in ' . $path);
+		}
+
+		if (is_array($value) && $this->metadataReader->isMessage($typeName)) {
+			return $this->denormalize($typeName, $value);
+		}
+
+		throw new InvalidMessageException('Unsupported type ' . $typeName . ' in ' . $path);
+	}
+}
diff --git a/lib/private/MessageQueue/MessageStore.php b/lib/private/MessageQueue/MessageStore.php
new file mode 100644
index 0000000000000..86b3ccd5d3e8e
--- /dev/null
+++ b/lib/private/MessageQueue/MessageStore.php
@@ -0,0 +1,223 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OC\MessageQueue;
+
+use OCP\AppFramework\Utility\ITimeFactory;
+use OCP\DB\QueryBuilder\ConflictResolutionMode;
+use OCP\DB\QueryBuilder\IQueryBuilder;
+use OCP\IDBConnection;
+use OCP\MessageQueue\Queue;
+use OCP\Snowflake\ISnowflakeGenerator;
+
+/**
+ * Stores messages in the message_queue table.
+ */
+final readonly class MessageStore {
+	public const string TABLE = 'message_queue';
+
+	private const string FAILED_QUEUE = 'failed';
+
+	/** Seconds after which a message claimed by a crashed worker is delivered again */
+	private const int REDELIVER_TIMEOUT = 3600;
+
+	private const int MAX_CLAIM_ATTEMPTS = 5;
+
+	public function __construct(
+		private IDBConnection $connection,
+		private ITimeFactory $timeFactory,
+		private ISnowflakeGenerator $snowflakeGenerator,
+	) {
+	}
+
+	/**
+	 * @param string|null $deduplicationHash Skips the insert while a message with this hash is pending
+	 * @return string|null Id of the stored message, null if it was deduplicated
+	 */
+	public function add(Queue $queue, string $messageClass, string $body, ?string $deduplicationHash): ?string {
+		if ($deduplicationHash !== null && $this->hasPending($queue, $deduplicationHash)) {
+			return null;
+		}
+
+		$now = $this->timeFactory->getTime();
+		$id = $this->snowflakeGenerator->nextId();
+		$qb = $this->connection->getQueryBuilder();
+		$qb->insert(self::TABLE)
+			->values([
+				'id' => $qb->createNamedParameter($id, IQueryBuilder::PARAM_INT),
+				'queue_name' => $qb->createNamedParameter($queue->value),
+				'message_class' => $qb->createNamedParameter($messageClass),
+				'body' => $qb->createNamedParameter($body),
+				'created_at' => $qb->createNamedParameter($now, IQueryBuilder::PARAM_INT),
+				'available_at' => $qb->createNamedParameter($now, IQueryBuilder::PARAM_INT),
+				'deduplication_hash' => $qb->createNamedParameter($deduplicationHash),
+			])
+			->executeStatement();
+		return $id;
+	}
+
+	/**
+	 * Claims the next available message of the queue.
+	 *
+	 * SQLite doesn't support SKIP LOCKED and Oracle can't combine it with a row
+	 * limit, so they claim optimistically and retry when another worker was faster.
+	 */
+	public function claimNext(Queue $queue): ?QueuedMessage {
+		$provider = $this->connection->getDatabaseProvider();
+		if ($provider === IDBConnection::PLATFORM_SQLITE || $provider === IDBConnection::PLATFORM_ORACLE) {
+			return $this->claimNextOptimistically($queue);
+		}
+
+		return $this->claimNextWithSkipLocked($queue);
+	}
+
+	/**
+	 * Locks the next available message, skipping the ones locked by other workers.
+	 */
+	private function claimNextWithSkipLocked(Queue $queue): ?QueuedMessage {
+		$this->connection->beginTransaction();
+		try {
+			$message = $this->fetchAvailable($queue, skipLocked: true);
+			$claimed = $message instanceof QueuedMessage && $this->claim($message);
+			$this->connection->commit();
+		} catch (\Throwable $throwable) {
+			$this->connection->rollBack();
+			throw $throwable;
+		}
+
+		return $claimed ? $message : null;
+	}
+
+	private function claimNextOptimistically(Queue $queue): ?QueuedMessage {
+		for ($attempt = 0; $attempt < self::MAX_CLAIM_ATTEMPTS; ++$attempt) {
+			$message = $this->fetchAvailable($queue);
+			if (!$message instanceof QueuedMessage) {
+				return null;
+			}
+
+			if ($this->claim($message)) {
+				return $message;
+			}
+		}
+
+		return null;
+	}
+
+	/**
+	 * Claims a specific message, if it is available.
+	 */
+	public function claimById(string $id): ?QueuedMessage {
+		$message = $this->fetchAvailable(id: $id);
+		return $message instanceof QueuedMessage && $this->claim($message) ? $message : null;
+	}
+
+	public function delete(string $id): void {
+		$qb = $this->connection->getQueryBuilder();
+		$qb->delete(self::TABLE)
+			->where($qb->expr()->eq('id', $qb->createNamedParameter($id, IQueryBuilder::PARAM_INT)))
+			->executeStatement();
+	}
+
+	/**
+	 * Makes a claimed message available again after the given delay.
+	 */
+	public function retry(string $id, int $delay): void {
+		$qb = $this->connection->getQueryBuilder();
+		$qb->update(self::TABLE)
+			->set('retry_count', $qb->func()->add('retry_count', $qb->createNamedParameter(1, IQueryBuilder::PARAM_INT)))
+			->set('available_at', $qb->createNamedParameter($this->timeFactory->getTime() + $delay, IQueryBuilder::PARAM_INT))
+			->set('delivered_at', $qb->createNamedParameter(null))
+			->where($qb->expr()->eq('id', $qb->createNamedParameter($id, IQueryBuilder::PARAM_INT)))
+			->executeStatement();
+	}
+
+	/**
+	 * Moves a claimed message to the failed queue.
+	 */
+	public function fail(string $id, string $error): void {
+		$qb = $this->connection->getQueryBuilder();
+		$qb->update(self::TABLE)
+			->set('queue_name', $qb->createNamedParameter(self::FAILED_QUEUE))
+			->set('last_error', $qb->createNamedParameter($error))
+			->set('delivered_at', $qb->createNamedParameter(null))
+			->set('deduplication_hash', $qb->createNamedParameter(null))
+			->where($qb->expr()->eq('id', $qb->createNamedParameter($id, IQueryBuilder::PARAM_INT)))
+			->executeStatement();
+	}
+
+	private function fetchAvailable(?Queue $queue = null, ?string $id = null, bool $skipLocked = false): ?QueuedMessage {
+		$now = $this->timeFactory->getTime();
+		$qb = $this->connection->getQueryBuilder();
+		$qb->select('id', 'message_class', 'body', 'retry_count', 'delivered_at')
+			->from(self::TABLE)
+			->where($qb->expr()->lte('available_at', $qb->createNamedParameter($now, IQueryBuilder::PARAM_INT)))
+			->andWhere($qb->expr()->orX(
+				$qb->expr()->isNull('delivered_at'),
+				$qb->expr()->lt('delivered_at', $qb->createNamedParameter($now - self::REDELIVER_TIMEOUT, IQueryBuilder::PARAM_INT)),
+			))
+			->orderBy('available_at', 'ASC')
+			->addOrderBy('id', 'ASC')
+			->setMaxResults(1);
+		if ($queue instanceof Queue) {
+			$qb->andWhere($qb->expr()->eq('queue_name', $qb->createNamedParameter($queue->value)));
+		}
+
+		if ($id !== null) {
+			$qb->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($id, IQueryBuilder::PARAM_INT)));
+		}
+
+		if ($skipLocked) {
+			$qb->forUpdate(ConflictResolutionMode::SkipLocked);
+		}
+
+		$result = $qb->executeQuery();
+		$row = $result->fetchAssociative();
+		$result->closeCursor();
+		if ($row === false) {
+			return null;
+		}
+
+		return new QueuedMessage(
+			(string)$row['id'],
+			(string)$row['message_class'],
+			(string)$row['body'],
+			(int)$row['retry_count'],
+			$row['delivered_at'] === null ? null : (int)$row['delivered_at'],
+		);
+	}
+
+	/**
+	 * Marks the message as delivered, unless another worker claimed it first.
+	 */
+	private function claim(QueuedMessage $message): bool {
+		$qb = $this->connection->getQueryBuilder();
+		$qb->update(self::TABLE)
+			->set('delivered_at', $qb->createNamedParameter($this->timeFactory->getTime(), IQueryBuilder::PARAM_INT))
+			->where($qb->expr()->eq('id', $qb->createNamedParameter($message->id, IQueryBuilder::PARAM_INT)));
+		if ($message->deliveredAt === null) {
+			$qb->andWhere($qb->expr()->isNull('delivered_at'));
+		} else {
+			$qb->andWhere($qb->expr()->eq('delivered_at', $qb->createNamedParameter($message->deliveredAt, IQueryBuilder::PARAM_INT)));
+		}
+
+		return $qb->executeStatement() === 1;
+	}
+
+	private function hasPending(Queue $queue, string $deduplicationHash): bool {
+		$qb = $this->connection->getQueryBuilder();
+		$qb->select('id')
+			->from(self::TABLE)
+			->where($qb->expr()->eq('deduplication_hash', $qb->createNamedParameter($deduplicationHash)))
+			->andWhere($qb->expr()->eq('queue_name', $qb->createNamedParameter($queue->value)))
+			->andWhere($qb->expr()->isNull('delivered_at'))
+			->setMaxResults(1);
+		$result = $qb->executeQuery();
+		$found = $result->fetchOne() !== false;
+		$result->closeCursor();
+		return $found;
+	}
+}
diff --git a/lib/private/MessageQueue/QueuedMessage.php b/lib/private/MessageQueue/QueuedMessage.php
new file mode 100644
index 0000000000000..61d8f4c30cbae
--- /dev/null
+++ b/lib/private/MessageQueue/QueuedMessage.php
@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OC\MessageQueue;
+
+/**
+ * A message claimed from the store, not yet decoded.
+ */
+final readonly class QueuedMessage {
+	public function __construct(
+		public string $id,
+		public string $messageClass,
+		public string $body,
+		public int $retryCount,
+		/** Previous delivery time, used to claim the message atomically */
+		public ?int $deliveredAt,
+	) {
+	}
+}
diff --git a/lib/private/Server.php b/lib/private/Server.php
index 77475ccc610f7..6108c290e1859 100644
--- a/lib/private/Server.php
+++ b/lib/private/Server.php
@@ -95,6 +95,7 @@
 use OC\Memcache\Factory;
 use OC\Memcache\NullCache;
 use OC\Memcache\Redis;
+use OC\MessageQueue\MessageBus;
 use OC\Notification\Manager;
 use OC\OCM\Model\OCMProvider;
 use OC\OCM\OCMDiscoveryService;
@@ -249,6 +250,7 @@
 use OCP\Log\ILogFactory;
 use OCP\Mail\IEmailValidator;
 use OCP\Mail\IMailer;
+use OCP\MessageQueue\IMessageBus;
 use OCP\OCM\ICapabilityAwareOCMProvider;
 use OCP\OCM\IOCMDiscoveryService;
 use OCP\OCM\IOCMProvider;
@@ -589,6 +591,7 @@ public function __construct(
 
 		$this->registerAlias(ILogFactory::class, LogFactory::class);
 		$this->registerAlias(IJobList::class, JobList::class);
+		$this->registerAlias(IMessageBus::class, MessageBus::class);
 
 		$this->registerService(Router::class, static function (Server $c) {
 			$cacheFactory = $c->get(ICacheFactory::class);
diff --git a/lib/public/App/AppInfoDefinition.php b/lib/public/App/AppInfoDefinition.php
index 261057b2f770d..5db4218441cc4 100644
--- a/lib/public/App/AppInfoDefinition.php
+++ b/lib/public/App/AppInfoDefinition.php
@@ -248,6 +248,7 @@
  *     'repair-steps'?: AppInfoFieldTypeRepairSteps,
  *     'two-factor-providers'?: list<class-string>,
  *     'commands'?: list<class-string>,
+ *     'message-handlers'?: list<class-string>,
  *     'settings'?: AppInfoFieldTypeSettings,
  *     'activity'?: AppInfoFieldTypeActivity,
  *     'dashboard'?: AppInfoFieldTypeDashboard,
diff --git a/lib/public/MessageQueue/Attribute/AsMessage.php b/lib/public/MessageQueue/Attribute/AsMessage.php
new file mode 100644
index 0000000000000..3ebe7018af19e
--- /dev/null
+++ b/lib/public/MessageQueue/Attribute/AsMessage.php
@@ -0,0 +1,60 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OCP\MessageQueue\Attribute;
+
+use OCP\AppFramework\Attribute\Consumable;
+use OCP\MessageQueue\Queue;
+
+/**
+ * Declares a class as a message that can be dispatched to the message queue.
+ *
+ * Messages are immutable data objects. They are serialized from their public
+ * constructor-promoted properties, which may only be of type int, float,
+ * string, bool, null, array (containing only those types), backed enums,
+ * \DateTimeImmutable or another class marked with #[AsMessage].
+ *
+ * Load entities and services inside the handler, never pass them in the message.
+ *
+ * ```
+ * #[AsMessage(queue: Queue::Low, deduplicateBy: ['fileId'], userProperty: 'userId')]
+ * final readonly class GeneratePreview {
+ *     public function __construct(
+ *         public int $fileId,
+ *         public string $userId,
+ *     ) {
+ *     }
+ * }
+ * ```
+ *
+ * @since 36.0.0
+ */
+#[\Attribute(\Attribute::TARGET_CLASS)]
+#[Consumable(since: '36.0.0')]
+final class AsMessage {
+	/**
+	 * @param Queue $queue Queue the message is routed to
+	 * @param int $maxRetries How often a failed message is retried before it is moved to the failure queue
+	 * @param int $retryDelay Delay in seconds before the first retry
+	 * @param float $retryMultiplier Factor applied to the delay after each retry
+	 * @param list<string> $deduplicateBy Properties forming the deduplication key. While a message
+	 *                                    with the same key is pending, dispatching it again is a no-op.
+	 *                                    An empty list disables deduplication.
+	 * @param string|null $userProperty Property holding a user id. The filesystem and user scope of
+	 *                                  that user are set up before the handler runs and torn down after.
+	 * @since 36.0.0
+	 */
+	public function __construct(
+		public Queue $queue = Queue::Default,
+		public int $maxRetries = 3,
+		public int $retryDelay = 60,
+		public float $retryMultiplier = 2.0,
+		public array $deduplicateBy = [],
+		public ?string $userProperty = null,
+	) {
+	}
+}
diff --git a/lib/public/MessageQueue/Attribute/AsMessageHandler.php b/lib/public/MessageQueue/Attribute/AsMessageHandler.php
new file mode 100644
index 0000000000000..50dc0541f983f
--- /dev/null
+++ b/lib/public/MessageQueue/Attribute/AsMessageHandler.php
@@ -0,0 +1,55 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OCP\MessageQueue\Attribute;
+
+use OCP\AppFramework\Attribute\Consumable;
+
+/**
+ * Declares the handler of a message. The handled message class is taken from
+ * the type of the first parameter. Each message class has exactly one handler.
+ *
+ * Handler classes are listed in appinfo/info.xml under <message-handlers> and
+ * are resolved through the app container, so constructor injection works.
+ *
+ * Can be used either on an invokable class:
+ *
+ * ```
+ * #[AsMessageHandler]
+ * final class GeneratePreviewHandler {
+ *     public function __construct(private IPreview $preview) {
+ *     }
+ *
+ *     public function __invoke(GeneratePreview $message): void {
+ *         // ...
+ *     }
+ * }
+ * ```
+ *
+ * Or on methods:
+ *
+ * ```
+ * final class FileTasks {
+ *     #[AsMessageHandler]
+ *     public function preview(GeneratePreview $message): void {
+ *     }
+ *
+ *     #[AsMessageHandler]
+ *     public function scan(ScanFolder $message): void {
+ *     }
+ * }
+ * ```
+ *
+ * Throw an UnrecoverableMessageException to fail the message without retrying.
+ * Any other exception triggers a retry according to the message's retry settings.
+ *
+ * @since 36.0.0
+ */
+#[\Attribute(\Attribute::TARGET_CLASS | \Attribute::TARGET_METHOD)]
+#[Consumable(since: '36.0.0')]
+final class AsMessageHandler {
+}
diff --git a/lib/public/MessageQueue/Exception/InvalidMessageException.php b/lib/public/MessageQueue/Exception/InvalidMessageException.php
new file mode 100644
index 0000000000000..7955503a493fd
--- /dev/null
+++ b/lib/public/MessageQueue/Exception/InvalidMessageException.php
@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OCP\MessageQueue\Exception;
+
+use OCP\AppFramework\Attribute\Catchable;
+
+/**
+ * Thrown on dispatch when a message cannot be queued.
+ *
+ * @since 36.0.0
+ */
+#[Catchable(since: '36.0.0')]
+final class InvalidMessageException extends \InvalidArgumentException {
+}
diff --git a/lib/public/MessageQueue/Exception/RecoverableMessageException.php b/lib/public/MessageQueue/Exception/RecoverableMessageException.php
new file mode 100644
index 0000000000000..485981241a6f0
--- /dev/null
+++ b/lib/public/MessageQueue/Exception/RecoverableMessageException.php
@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OCP\MessageQueue\Exception;
+
+use OCP\AppFramework\Attribute\Throwable;
+
+/**
+ * Thrown by a handler to always retry a message, even when maxRetries is reached,
+ * e.g. when a remote service is temporarily unavailable.
+ *
+ * @since 36.0.0
+ */
+#[Throwable(since: '36.0.0')]
+final class RecoverableMessageException extends \RuntimeException {
+}
diff --git a/lib/public/MessageQueue/Exception/UnrecoverableMessageException.php b/lib/public/MessageQueue/Exception/UnrecoverableMessageException.php
new file mode 100644
index 0000000000000..b3717ae345ad0
--- /dev/null
+++ b/lib/public/MessageQueue/Exception/UnrecoverableMessageException.php
@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OCP\MessageQueue\Exception;
+
+use OCP\AppFramework\Attribute\Throwable;
+
+/**
+ * Thrown by a handler to fail a message permanently, skipping remaining retries.
+ *
+ * @since 36.0.0
+ */
+#[Throwable(since: '36.0.0')]
+final class UnrecoverableMessageException extends \RuntimeException {
+}
diff --git a/lib/public/MessageQueue/IMessageBus.php b/lib/public/MessageQueue/IMessageBus.php
new file mode 100644
index 0000000000000..a6bacfbaa912f
--- /dev/null
+++ b/lib/public/MessageQueue/IMessageBus.php
@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OCP\MessageQueue;
+
+use OCP\AppFramework\Attribute\Consumable;
+use OCP\MessageQueue\Exception\InvalidMessageException;
+
+/**
+ * Dispatches messages to be handled asynchronously.
+ *
+ * Messages are stored in the database. When running under PHP-FPM, messages
+ * dispatched during a web request are handled right after the response was
+ * sent, except for those in the Low queue. Remaining messages are consumed
+ * during the regular cron run. Large instances can run dedicated
+ * `occ message-queue:consume` workers.
+ *
+ * @since 36.0.0
+ */
+#[Consumable(since: '36.0.0')]
+interface IMessageBus {
+	/**
+	 * @param object $message Instance of a class marked with #[AsMessage]
+	 * @throws InvalidMessageException if the message is not marked with #[AsMessage],
+	 *                                 has no handler or contains unsupported property types
+	 * @since 36.0.0
+	 */
+	public function dispatch(object $message): void;
+}
diff --git a/lib/public/MessageQueue/Queue.php b/lib/public/MessageQueue/Queue.php
new file mode 100644
index 0000000000000..b24e5a04fc992
--- /dev/null
+++ b/lib/public/MessageQueue/Queue.php
@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace OCP\MessageQueue;
+
+use OCP\AppFramework\Attribute\Consumable;
+
+/**
+ * Cron consumes queues in order of priority: High before Default before Low.
+ * Workers consume the queues passed to `occ message-queue:consume` in the given order.
+ *
+ * @since 36.0.0
+ */
+#[Consumable(since: '36.0.0')]
+enum Queue: string {
+	/**
+	 * Interactive work a user is waiting for.
+	 * @since 36.0.0
+	 */
+	case High = 'high';
+	/**
+	 * @since 36.0.0
+	 */
+	case Default = 'default';
+	/**
+	 * Expensive work. When a maintenance window is configured, system cron
+	 * only consumes it during that window. AJAX and webcron always consume it.
+	 * @since 36.0.0
+	 */
+	case Low = 'low';
+}
diff --git a/psalm-strict.xml b/psalm-strict.xml
index 383d80f2d8b49..8bd90cb511a97 100644
--- a/psalm-strict.xml
+++ b/psalm-strict.xml
@@ -69,6 +69,10 @@
 		<directory name="lib/public/SystemReport"/>
 		<directory name="lib/private/SystemReport"/>
 		<directory name="tests/lib/SystemReport"/>
+		<directory name="lib/public/MessageQueue"/>
+		<directory name="lib/private/MessageQueue"/>
+		<directory name="core/Command/MessageQueue"/>
+		<directory name="tests/lib/MessageQueue"/>
 		<ignoreFiles>
 			<!-- Circles internals are used -->
 			<file name="core/Sharing/Recipient/TeamShareRecipientType.php"/>
diff --git a/resources/app-info-shipped.xsd b/resources/app-info-shipped.xsd
index c7f3808ca9dd9..7f910d7b0b354 100644
--- a/resources/app-info-shipped.xsd
+++ b/resources/app-info-shipped.xsd
@@ -55,6 +55,8 @@
                             minOccurs="0" maxOccurs="1"/>
                 <xs:element name="commands" type="commands"
                             minOccurs="0" maxOccurs="1"/>
+                <xs:element name="message-handlers" type="message-handlers"
+                            minOccurs="0" maxOccurs="1"/>
                 <xs:element name="settings" type="settings" minOccurs="0"
                             maxOccurs="1"/>
                 <xs:element name="activity" type="activity" minOccurs="0"
@@ -804,6 +806,13 @@
         </xs:sequence>
     </xs:complexType>
 
+    <xs:complexType name="message-handlers">
+        <xs:sequence>
+            <xs:element name="handler" type="php-class" minOccurs="1"
+                        maxOccurs="unbounded"/>
+        </xs:sequence>
+    </xs:complexType>
+
     <xs:complexType name="dashboard">
         <xs:sequence>
             <xs:element name="widget" type="php-class" minOccurs="1"
diff --git a/resources/app-info.xsd b/resources/app-info.xsd
index 0889c7d431500..cd59d8d5ec220 100644
--- a/resources/app-info.xsd
+++ b/resources/app-info.xsd
@@ -53,6 +53,8 @@
                             minOccurs="0" maxOccurs="1"/>
                 <xs:element name="commands" type="commands"
                             minOccurs="0" maxOccurs="1"/>
+                <xs:element name="message-handlers" type="message-handlers"
+                            minOccurs="0" maxOccurs="1"/>
                 <xs:element name="settings" type="settings" minOccurs="0"
                             maxOccurs="1"/>
                 <xs:element name="activity" type="activity" minOccurs="0"
@@ -784,6 +786,13 @@
         </xs:sequence>
     </xs:complexType>
 
+    <xs:complexType name="message-handlers">
+        <xs:sequence>
+            <xs:element name="handler" type="php-class" minOccurs="1"
+                        maxOccurs="unbounded"/>
+        </xs:sequence>
+    </xs:complexType>
+
     <xs:complexType name="dashboard">
         <xs:sequence>
             <xs:element name="widget" type="php-class" minOccurs="1"
diff --git a/tests/data/app/appinfo-attributes-once.json b/tests/data/app/appinfo-attributes-once.json
index 3ecc803733d54..75ea195d1d856 100644
--- a/tests/data/app/appinfo-attributes-once.json
+++ b/tests/data/app/appinfo-attributes-once.json
@@ -99,6 +99,7 @@
     "commands": {
         "command": "OCA\\AttributesOnce\\Command\\Run"
     },
+    "message-handlers": [],
     "settings": {
         "admin": [
             "OCA\\AttributesOnce\\Settings\\Admin"
diff --git a/tests/data/app/appinfo-multi-once.json b/tests/data/app/appinfo-multi-once.json
index 0145060ebe7c9..da9c683c2b74f 100644
--- a/tests/data/app/appinfo-multi-once.json
+++ b/tests/data/app/appinfo-multi-once.json
@@ -74,6 +74,7 @@
     "commands": {
         "command": "OCA\\MultiOnce\\Command\\Migrate"
     },
+    "message-handlers": [],
     "settings": {
         "admin": [
             "OCA\\MultiOnce\\Settings\\Admin"
diff --git a/tests/data/app/appinfo-multi-twice.json b/tests/data/app/appinfo-multi-twice.json
index 114cab05f6c2f..845976f80a171 100644
--- a/tests/data/app/appinfo-multi-twice.json
+++ b/tests/data/app/appinfo-multi-twice.json
@@ -135,6 +135,7 @@
         "OCA\\MultiTwice\\Command\\MigrateOne",
         "OCA\\MultiTwice\\Command\\MigrateTwo"
     ],
+    "message-handlers": [],
     "settings": {
         "admin": [
             "OCA\\MultiTwice\\Settings\\AdminOne",
diff --git a/tests/data/app/expected-info.json b/tests/data/app/expected-info.json
index d34f9e67ed38d..f823a2542b2fc 100644
--- a/tests/data/app/expected-info.json
+++ b/tests/data/app/expected-info.json
@@ -80,6 +80,7 @@
 	"background-jobs": [],
 	"two-factor-providers": [],
 	"commands": [],
+	"message-handlers": [],
 	"activity": {
 		"filters": [],
 		"settings": [],
diff --git a/tests/data/app/navigation-one-item.json b/tests/data/app/navigation-one-item.json
index a9ce44f105df8..06e3756f2745b 100644
--- a/tests/data/app/navigation-one-item.json
+++ b/tests/data/app/navigation-one-item.json
@@ -39,6 +39,7 @@
   "commands": {
     "command": "OCA\\Activity\\Command\\SendEmails"
   },
+  "message-handlers": [],
   "settings": {
     "admin": [
       "OCA\\Activity\\Settings\\Admin"
diff --git a/tests/data/app/navigation-two-items.json b/tests/data/app/navigation-two-items.json
index 118c924279457..38f5e66d39e2a 100644
--- a/tests/data/app/navigation-two-items.json
+++ b/tests/data/app/navigation-two-items.json
@@ -39,6 +39,7 @@
   "commands": {
     "command": "OCA\\Activity\\Command\\SendEmails"
   },
+  "message-handlers": [],
   "settings": {
     "admin": [
       "OCA\\Activity\\Settings\\Admin"
diff --git a/tests/data/app/various-single-item.json b/tests/data/app/various-single-item.json
index ad845adc4b0a8..a707fe9392e4c 100644
--- a/tests/data/app/various-single-item.json
+++ b/tests/data/app/various-single-item.json
@@ -29,6 +29,7 @@
 		"providers": []
 	},
 	"commands": [],
+	"message-handlers": [],
 	"remote": [],
 	"public": [],
 	"repair-steps": {
diff --git a/tests/lib/MessageQueue/AfterRequestHandlerTest.php b/tests/lib/MessageQueue/AfterRequestHandlerTest.php
new file mode 100644
index 0000000000000..8af27b60c4456
--- /dev/null
+++ b/tests/lib/MessageQueue/AfterRequestHandlerTest.php
@@ -0,0 +1,92 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue;
+
+use OC\MessageQueue\Consumer;
+use OCP\Files\ISetupManager;
+use OCP\IConfig;
+use OCP\ISession;
+use OCP\IUserSession;
+use OCP\Lock\ILockingProvider;
+use OCP\MessageQueue\Queue;
+use Psr\Log\LoggerInterface;
+use Test\MessageQueue\Fixtures\TestAfterRequestHandler;
+use Test\TestCase;
+
+final class AfterRequestHandlerTest extends TestCase {
+	private function createHandler(
+		Consumer $consumer,
+		bool $enabled = true,
+		bool $isCLI = false,
+		?ILockingProvider $lockingProvider = null,
+		?LoggerInterface $logger = null,
+	): TestAfterRequestHandler {
+		$config = $this->createMock(IConfig::class);
+		$config->method('getSystemValueBool')->with('message_queue.handle_after_request', true)->willReturn($enabled);
+
+		return new TestAfterRequestHandler(
+			$consumer,
+			$config,
+			$this->createMock(ISession::class),
+			$this->createMock(IUserSession::class),
+			$this->createMock(ISetupManager::class),
+			$lockingProvider ?? $this->createMock(ILockingProvider::class),
+			$logger ?? $this->createMock(LoggerInterface::class),
+			$isCLI,
+		);
+	}
+
+	public function testEnabled(): void {
+		$consumer = $this->createMock(Consumer::class);
+		$consumer->expects($this->once())->method('consumeMessages')->with(['1', '2', '3'], 10, 5);
+
+		$handler = $this->createHandler($consumer);
+		$handler->add(Queue::High, '1');
+		$handler->add(Queue::Default, '2');
+		$handler->add(Queue::High, '3');
+		$handler->add(Queue::Low, '4');
+		$this->assertCount(1, $handler->shutdownFunctions);
+
+		$handler->run();
+		$handler->run();
+		$this->assertTrue($handler->requestFinished);
+	}
+
+	public function testDisabledOnCli(): void {
+		$consumer = $this->createMock(Consumer::class);
+		$consumer->expects($this->never())->method('consumeMessages');
+
+		$handler = $this->createHandler($consumer, isCLI: true);
+		$handler->add(Queue::High, '1');
+		$handler->run();
+		$this->assertSame([], $handler->shutdownFunctions);
+	}
+
+	public function testDisabledByConfig(): void {
+		$consumer = $this->createMock(Consumer::class);
+		$consumer->expects($this->never())->method('consumeMessages');
+
+		$handler = $this->createHandler($consumer, enabled: false);
+		$handler->add(Queue::High, '1');
+		$handler->run();
+		$this->assertSame([], $handler->shutdownFunctions);
+	}
+
+	public function testErrorsAreLoggedAndLocksReleased(): void {
+		$consumer = $this->createMock(Consumer::class);
+		$consumer->method('consumeMessages')->willThrowException(new \RuntimeException('Broken'));
+		$logger = $this->createMock(LoggerInterface::class);
+		$logger->expects($this->once())->method('error');
+		$lockingProvider = $this->createMock(ILockingProvider::class);
+		$lockingProvider->expects($this->once())->method('releaseAll');
+
+		$handler = $this->createHandler($consumer, lockingProvider: $lockingProvider, logger: $logger);
+		$handler->add(Queue::High, '1');
+		$handler->run();
+	}
+}
diff --git a/tests/lib/MessageQueue/Fixtures/DuplicateHandler.php b/tests/lib/MessageQueue/Fixtures/DuplicateHandler.php
new file mode 100644
index 0000000000000..d21520e8e2bed
--- /dev/null
+++ b/tests/lib/MessageQueue/Fixtures/DuplicateHandler.php
@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue\Fixtures;
+
+use OCP\MessageQueue\Attribute\AsMessageHandler;
+
+final class DuplicateHandler {
+	#[AsMessageHandler]
+	public function duplicate(TestMessage $message): void {
+	}
+
+	#[AsMessageHandler]
+	public function invalid(NotAMessage $message): void {
+	}
+}
diff --git a/tests/lib/MessageQueue/Fixtures/FailingMessage.php b/tests/lib/MessageQueue/Fixtures/FailingMessage.php
new file mode 100644
index 0000000000000..8aeabadf87c82
--- /dev/null
+++ b/tests/lib/MessageQueue/Fixtures/FailingMessage.php
@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue\Fixtures;
+
+use OCP\MessageQueue\Attribute\AsMessage;
+
+#[AsMessage(maxRetries: 1, retryDelay: 30)]
+final readonly class FailingMessage {
+	public function __construct(
+		public bool $unrecoverable = false,
+		public bool $recoverable = false,
+	) {
+	}
+}
diff --git a/tests/lib/MessageQueue/Fixtures/InvokableHandler.php b/tests/lib/MessageQueue/Fixtures/InvokableHandler.php
new file mode 100644
index 0000000000000..0a31986a74007
--- /dev/null
+++ b/tests/lib/MessageQueue/Fixtures/InvokableHandler.php
@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue\Fixtures;
+
+use OCP\MessageQueue\Attribute\AsMessageHandler;
+
+#[AsMessageHandler]
+final class InvokableHandler {
+	/** @var list<TestMessage> */
+	public array $handled = [];
+
+	public function __invoke(TestMessage $message): void {
+		$this->handled[] = $message;
+	}
+}
diff --git a/tests/lib/MessageQueue/Fixtures/LowMessage.php b/tests/lib/MessageQueue/Fixtures/LowMessage.php
new file mode 100644
index 0000000000000..02c17a9deb2fe
--- /dev/null
+++ b/tests/lib/MessageQueue/Fixtures/LowMessage.php
@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue\Fixtures;
+
+use OCP\MessageQueue\Attribute\AsMessage;
+use OCP\MessageQueue\Queue;
+
+#[AsMessage(queue: Queue::Low)]
+final readonly class LowMessage {
+	public function __construct(
+		public int $id,
+	) {
+	}
+}
diff --git a/tests/lib/MessageQueue/Fixtures/MultiHandler.php b/tests/lib/MessageQueue/Fixtures/MultiHandler.php
new file mode 100644
index 0000000000000..73b647022892d
--- /dev/null
+++ b/tests/lib/MessageQueue/Fixtures/MultiHandler.php
@@ -0,0 +1,52 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue\Fixtures;
+
+use OCP\IUserSession;
+use OCP\MessageQueue\Attribute\AsMessageHandler;
+use OCP\MessageQueue\Exception\RecoverableMessageException;
+use OCP\MessageQueue\Exception\UnrecoverableMessageException;
+
+final class MultiHandler {
+	/** @var list<string|null> */
+	public array $users = [];
+
+	/** @var list<LowMessage> */
+	public array $low = [];
+
+	public function __construct(
+		private readonly IUserSession $userSession,
+	) {
+	}
+
+	#[AsMessageHandler]
+	public function user(UserMessage $message): void {
+		$this->users[] = $this->userSession->getUser()?->getUID();
+	}
+
+	#[AsMessageHandler]
+	public function fail(FailingMessage $message): void {
+		if ($message->unrecoverable) {
+			throw new UnrecoverableMessageException('Permanent failure');
+		}
+
+		if ($message->recoverable) {
+			throw new RecoverableMessageException('Service unavailable');
+		}
+
+		throw new \RuntimeException('Temporary failure');
+	}
+
+	#[AsMessageHandler]
+	public function low(LowMessage $message): void {
+		$this->low[] = $message;
+	}
+
+	public function notAHandler(TestMessage $message): void {
+	}
+}
diff --git a/tests/lib/MessageQueue/Fixtures/NestedMessage.php b/tests/lib/MessageQueue/Fixtures/NestedMessage.php
new file mode 100644
index 0000000000000..44f89e63d1626
--- /dev/null
+++ b/tests/lib/MessageQueue/Fixtures/NestedMessage.php
@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue\Fixtures;
+
+use OCP\MessageQueue\Attribute\AsMessage;
+
+#[AsMessage]
+final readonly class NestedMessage {
+	public function __construct(
+		public string $name,
+	) {
+	}
+}
diff --git a/tests/lib/MessageQueue/Fixtures/NotAMessage.php b/tests/lib/MessageQueue/Fixtures/NotAMessage.php
new file mode 100644
index 0000000000000..401d845f83781
--- /dev/null
+++ b/tests/lib/MessageQueue/Fixtures/NotAMessage.php
@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue\Fixtures;
+
+final readonly class NotAMessage {
+	public function __construct(
+		public int $id,
+	) {
+	}
+}
diff --git a/tests/lib/MessageQueue/Fixtures/Status.php b/tests/lib/MessageQueue/Fixtures/Status.php
new file mode 100644
index 0000000000000..4f1c7303aad2f
--- /dev/null
+++ b/tests/lib/MessageQueue/Fixtures/Status.php
@@ -0,0 +1,13 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue\Fixtures;
+
+enum Status: string {
+	case Active = 'active';
+	case Disabled = 'disabled';
+}
diff --git a/tests/lib/MessageQueue/Fixtures/TestAfterRequestHandler.php b/tests/lib/MessageQueue/Fixtures/TestAfterRequestHandler.php
new file mode 100644
index 0000000000000..498ed22a073ec
--- /dev/null
+++ b/tests/lib/MessageQueue/Fixtures/TestAfterRequestHandler.php
@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue\Fixtures;
+
+use OC\MessageQueue\AfterRequestHandler;
+
+/**
+ * Records shutdown functions and the end of the request instead of acting on them.
+ */
+final class TestAfterRequestHandler extends AfterRequestHandler {
+	/** @var list<callable> */
+	public array $shutdownFunctions = [];
+
+	public bool $requestFinished = false;
+
+	#[\Override]
+	protected function canFinishRequest(): bool {
+		return true;
+	}
+
+	#[\Override]
+	protected function registerShutdownFunction(callable $callback): void {
+		$this->shutdownFunctions[] = $callback;
+	}
+
+	#[\Override]
+	protected function finishRequest(): void {
+		$this->requestFinished = true;
+	}
+}
diff --git a/tests/lib/MessageQueue/Fixtures/TestMessage.php b/tests/lib/MessageQueue/Fixtures/TestMessage.php
new file mode 100644
index 0000000000000..39820b9d1be40
--- /dev/null
+++ b/tests/lib/MessageQueue/Fixtures/TestMessage.php
@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue\Fixtures;
+
+use OCP\MessageQueue\Attribute\AsMessage;
+use OCP\MessageQueue\Queue;
+
+#[AsMessage(queue: Queue::High, maxRetries: 2, retryDelay: 10, retryMultiplier: 3.0, deduplicateBy: ['id'])]
+final readonly class TestMessage {
+	public function __construct(
+		public int $id,
+		public string $text = 'default',
+		public ?Status $status = null,
+		public ?\DateTimeImmutable $date = null,
+		public ?NestedMessage $nested = null,
+		public array $list = [],
+		public float $ratio = 1.0,
+	) {
+	}
+}
diff --git a/tests/lib/MessageQueue/Fixtures/UserMessage.php b/tests/lib/MessageQueue/Fixtures/UserMessage.php
new file mode 100644
index 0000000000000..5b82183b69d46
--- /dev/null
+++ b/tests/lib/MessageQueue/Fixtures/UserMessage.php
@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue\Fixtures;
+
+use OCP\MessageQueue\Attribute\AsMessage;
+
+#[AsMessage(userProperty: 'userId')]
+final readonly class UserMessage {
+	public function __construct(
+		public string $userId,
+	) {
+	}
+}
diff --git a/tests/lib/MessageQueue/HandlerRegistryTest.php b/tests/lib/MessageQueue/HandlerRegistryTest.php
new file mode 100644
index 0000000000000..87384145c95e5
--- /dev/null
+++ b/tests/lib/MessageQueue/HandlerRegistryTest.php
@@ -0,0 +1,79 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue;
+
+use OC\Memcache\ArrayCache;
+use OC\MessageQueue\HandlerRegistry;
+use OC\MessageQueue\MessageMetadataReader;
+use OCP\App\IAppManager;
+use OCP\ICacheFactory;
+use PHPUnit\Framework\MockObject\MockObject;
+use Psr\Log\LoggerInterface;
+use Test\MessageQueue\Fixtures\DuplicateHandler;
+use Test\MessageQueue\Fixtures\FailingMessage;
+use Test\MessageQueue\Fixtures\InvokableHandler;
+use Test\MessageQueue\Fixtures\LowMessage;
+use Test\MessageQueue\Fixtures\MultiHandler;
+use Test\MessageQueue\Fixtures\NotAMessage;
+use Test\MessageQueue\Fixtures\TestMessage;
+use Test\MessageQueue\Fixtures\UserMessage;
+use Test\TestCase;
+
+final class HandlerRegistryTest extends TestCase {
+	private IAppManager&MockObject $appManager;
+
+	private LoggerInterface&MockObject $logger;
+
+	private ArrayCache $cache;
+
+	private HandlerRegistry $registry;
+
+	#[\Override]
+	protected function setUp(): void {
+		parent::setUp();
+		$this->appManager = $this->createMock(IAppManager::class);
+		$this->logger = $this->createMock(LoggerInterface::class);
+		$this->cache = new ArrayCache();
+		$cacheFactory = $this->createMock(ICacheFactory::class);
+		$cacheFactory->method('createLocal')->willReturn($this->cache);
+
+		$this->appManager->method('getEnabledApps')->willReturn(['app1', 'app2']);
+		$this->appManager->method('getAppVersion')->willReturn('1.0.0');
+		$this->appManager->method('getAppInfo')->willReturnMap([
+			['app1', false, null, ['message-handlers' => [InvokableHandler::class, MultiHandler::class]]],
+			['app2', false, null, ['message-handlers' => ['handler' => DuplicateHandler::class]]],
+		]);
+
+		$this->registry = new HandlerRegistry($this->appManager, new MessageMetadataReader(), $this->logger, $cacheFactory);
+	}
+
+	public function testCollectsHandlers(): void {
+		$this->logger->expects($this->exactly(2))->method('error');
+
+		$this->assertSame([
+			TestMessage::class => ['class' => InvokableHandler::class, 'method' => '__invoke'],
+			UserMessage::class => ['class' => MultiHandler::class, 'method' => 'user'],
+			FailingMessage::class => ['class' => MultiHandler::class, 'method' => 'fail'],
+			LowMessage::class => ['class' => MultiHandler::class, 'method' => 'low'],
+		], $this->registry->getHandlers());
+		$this->assertNull($this->registry->getHandler(NotAMessage::class));
+	}
+
+	public function testUsesCachedHandlers(): void {
+		$cached = [TestMessage::class => ['class' => InvokableHandler::class, 'method' => '__invoke']];
+		$this->cache->set('handlers-' . md5(json_encode(['app1' => '1.0.0', 'app2' => '1.0.0'], JSON_THROW_ON_ERROR)), $cached);
+		$this->appManager->expects($this->never())->method('getAppInfo');
+
+		$this->assertSame($cached, $this->registry->getHandlers());
+	}
+
+	public function testStoresHandlersInCache(): void {
+		$handlers = $this->registry->getHandlers();
+		$this->assertSame($handlers, $this->cache->get('handlers-' . md5(json_encode(['app1' => '1.0.0', 'app2' => '1.0.0'], JSON_THROW_ON_ERROR))));
+	}
+}
diff --git a/tests/lib/MessageQueue/MessageNormalizerTest.php b/tests/lib/MessageQueue/MessageNormalizerTest.php
new file mode 100644
index 0000000000000..c7a7f347df9c8
--- /dev/null
+++ b/tests/lib/MessageQueue/MessageNormalizerTest.php
@@ -0,0 +1,84 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue;
+
+use OC\MessageQueue\MessageMetadataReader;
+use OC\MessageQueue\MessageNormalizer;
+use OCP\MessageQueue\Exception\InvalidMessageException;
+use PHPUnit\Framework\Attributes\DataProvider;
+use Test\MessageQueue\Fixtures\NestedMessage;
+use Test\MessageQueue\Fixtures\Status;
+use Test\MessageQueue\Fixtures\TestMessage;
+use Test\TestCase;
+
+final class MessageNormalizerTest extends TestCase {
+	private MessageNormalizer $normalizer;
+
+	#[\Override]
+	protected function setUp(): void {
+		parent::setUp();
+		$this->normalizer = new MessageNormalizer(new MessageMetadataReader());
+	}
+
+	public function testRoundTrip(): void {
+		$message = new TestMessage(
+			id: 42,
+			text: 'hello',
+			status: Status::Disabled,
+			date: new \DateTimeImmutable('2026-10-04 12:34:56.123456+02:00'),
+			nested: new NestedMessage('child'),
+			list: ['a' => [1, 2], 'b' => null],
+			ratio: 2.5,
+		);
+
+		$data = $this->normalizer->normalize($message);
+		$this->assertSame([
+			'id' => 42,
+			'text' => 'hello',
+			'status' => 'disabled',
+			'date' => '2026-10-04T12:34:56.123456+02:00',
+			'nested' => ['name' => 'child'],
+			'list' => ['a' => [1, 2], 'b' => null],
+			'ratio' => 2.5,
+		], $data);
+
+		$json = json_decode(json_encode($data, JSON_THROW_ON_ERROR), true, flags: JSON_THROW_ON_ERROR);
+		$this->assertIsArray($json);
+		$this->assertEquals($message, $this->normalizer->denormalize(TestMessage::class, $json));
+	}
+
+	public function testDenormalizeUsesDefaultsForMissingValues(): void {
+		$message = $this->normalizer->denormalize(TestMessage::class, ['id' => 1, 'ratio' => 3]);
+		$this->assertEquals(new TestMessage(id: 1, ratio: 3.0), $message);
+	}
+
+	public function testNormalizeRejectsObjectsInArrays(): void {
+		$this->expectException(InvalidMessageException::class);
+		$this->normalizer->normalize(new TestMessage(id: 1, list: [Status::Active]));
+	}
+
+	/**
+	 * @return array<string, array{array<string, mixed>}>
+	 */
+	public static function invalidDataProvider(): array {
+		return [
+			'missing required value' => [['text' => 'a']],
+			'wrong scalar type' => [['id' => '1']],
+			'unexpected null' => [['id' => null]],
+			'invalid enum value' => [['id' => 1, 'status' => 'unknown']],
+			'invalid date' => [['id' => 1, 'date' => 'yesterday']],
+			'invalid nested message' => [['id' => 1, 'nested' => 'child']],
+		];
+	}
+
+	#[DataProvider('invalidDataProvider')]
+	public function testDenormalizeRejectsInvalidData(array $data): void {
+		$this->expectException(InvalidMessageException::class);
+		$this->normalizer->denormalize(TestMessage::class, $data);
+	}
+}
diff --git a/tests/lib/MessageQueue/MessageQueueTest.php b/tests/lib/MessageQueue/MessageQueueTest.php
new file mode 100644
index 0000000000000..ccdf9f8453b39
--- /dev/null
+++ b/tests/lib/MessageQueue/MessageQueueTest.php
@@ -0,0 +1,342 @@
+<?php
+
+declare(strict_types=1);
+
+// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+namespace Test\MessageQueue;
+
+use OC\MessageQueue\Consumer;
+use OC\MessageQueue\HandlerRegistry;
+use OC\MessageQueue\MessageBus;
+use OC\MessageQueue\MessageMetadataReader;
+use OC\MessageQueue\MessageNormalizer;
+use OC\MessageQueue\MessageStore;
+use OCP\AppFramework\Utility\ITimeFactory;
+use OCP\DB\QueryBuilder\IQueryBuilder;
+use OCP\Files\ISetupManager;
+use OCP\IConfig;
+use OCP\IDBConnection;
+use OCP\ISession;
+use OCP\ITempManager;
+use OCP\IUser;
+use OCP\IUserManager;
+use OCP\IUserSession;
+use OCP\Lock\ILockingProvider;
+use OCP\MessageQueue\Exception\InvalidMessageException;
+use OCP\MessageQueue\Exception\UnrecoverableMessageException;
+use OCP\Server;
+use OCP\Snowflake\ISnowflakeGenerator;
+use PHPUnit\Framework\Attributes\Group;
+use PHPUnit\Framework\MockObject\MockObject;
+use Psr\Container\ContainerInterface;
+use Psr\Log\NullLogger;
+use Test\MessageQueue\Fixtures\FailingMessage;
+use Test\MessageQueue\Fixtures\InvokableHandler;
+use Test\MessageQueue\Fixtures\LowMessage;
+use Test\MessageQueue\Fixtures\MultiHandler;
+use Test\MessageQueue\Fixtures\NotAMessage;
+use Test\MessageQueue\Fixtures\TestAfterRequestHandler;
+use Test\MessageQueue\Fixtures\TestMessage;
+use Test\MessageQueue\Fixtures\UserMessage;
+use Test\TestCase;
+
+#[Group('DB')]
+final class MessageQueueTest extends TestCase {
+	private IDBConnection $connection;
+
+	private int $now = 1_800_000_000;
+
+	private ?IUser $activeUser = null;
+
+	private ISetupManager&MockObject $setupManager;
+
+	private InvokableHandler $invokableHandler;
+
+	private MultiHandler $multiHandler;
+
+	private MessageBus $bus;
+
+	private Consumer $consumer;
+
+	private ISession&MockObject $session;
+
+	private ILockingProvider&MockObject $lockingProvider;
+
+	private TestAfterRequestHandler $afterRequestHandler;
+
+	#[\Override]
+	protected function setUp(): void {
+		parent::setUp();
+		$this->connection = Server::get(IDBConnection::class);
+		$this->clearTable();
+
+		$timeFactory = $this->createMock(ITimeFactory::class);
+		$timeFactory->method('getTime')->willReturnCallback(fn (): int => $this->now);
+
+		$userSession = $this->createMock(IUserSession::class);
+		$userSession->method('setVolatileActiveUser')->willReturnCallback(function (?IUser $user): void {
+			$this->activeUser = $user;
+		});
+		$userSession->method('getUser')->willReturnCallback(fn (): ?IUser => $this->activeUser);
+
+		$alice = $this->createMock(IUser::class);
+		$alice->method('getUID')->willReturn('alice');
+		$userManager = $this->createMock(IUserManager::class);
+		$userManager->method('get')->willReturnCallback(static fn (string $uid) => $uid === 'alice' ? $alice : null);
+
+		$this->setupManager = $this->createMock(ISetupManager::class);
+		$this->invokableHandler = new InvokableHandler();
+		$this->multiHandler = new MultiHandler($userSession);
+
+		$container = $this->createMock(ContainerInterface::class);
+		$container->method('get')->willReturnMap([
+			[InvokableHandler::class, $this->invokableHandler],
+			[MultiHandler::class, $this->multiHandler],
+		]);
+
+		$handlerRegistry = $this->createMock(HandlerRegistry::class);
+		$handlerRegistry->method('getHandler')->willReturnCallback(static fn (string $class): ?array => match ($class) {
+			TestMessage::class => ['class' => InvokableHandler::class, 'method' => '__invoke'],
+			UserMessage::class => ['class' => MultiHandler::class, 'method' => 'user'],
+			FailingMessage::class => ['class' => MultiHandler::class, 'method' => 'fail'],
+			LowMessage::class => ['class' => MultiHandler::class, 'method' => 'low'],
+			default => null,
+		});
+
+		$logger = new NullLogger();
+		$metadataReader = new MessageMetadataReader();
+		$normalizer = new MessageNormalizer($metadataReader);
+		$store = new MessageStore($this->connection, $timeFactory, Server::get(ISnowflakeGenerator::class));
+		$this->consumer = new Consumer(
+			$store,
+			$handlerRegistry,
+			$metadataReader,
+			$normalizer,
+			$container,
+			$userManager,
+			$userSession,
+			$this->setupManager,
+			$this->connection,
+			$this->createMock(ITempManager::class),
+			$logger,
+		);
+		$this->session = $this->createMock(ISession::class);
+		$this->lockingProvider = $this->createMock(ILockingProvider::class);
+		$config = $this->createMock(IConfig::class);
+		$config->method('getSystemValueBool')->willReturn(true);
+		$this->afterRequestHandler = new TestAfterRequestHandler($this->consumer, $config, $this->session, $userSession, $this->setupManager, $this->lockingProvider, $logger, false);
+		$this->bus = new MessageBus($store, $handlerRegistry, $metadataReader, $normalizer, $this->afterRequestHandler);
+	}
+
+	#[\Override]
+	protected function tearDown(): void {
+		$this->clearTable();
+		parent::tearDown();
+	}
+
+	private function clearTable(): void {
+		$this->connection->getQueryBuilder()->delete(MessageStore::TABLE)->executeStatement();
+	}
+
+	/**
+	 * @return list<array{queue_name: string, available_at: int, retry_count: int, last_error: ?string}>
+	 */
+	private function getRows(): array {
+		$qb = $this->connection->getQueryBuilder();
+		$result = $qb->select('queue_name', 'available_at', 'retry_count', 'last_error')
+			->from(MessageStore::TABLE)
+			->orderBy('id')
+			->executeQuery();
+		$rows = array_map(static fn (array $row): array => [
+			'queue_name' => (string)$row['queue_name'],
+			'available_at' => (int)$row['available_at'],
+			'retry_count' => (int)$row['retry_count'],
+			'last_error' => $row['last_error'] === null ? null : (string)$row['last_error'],
+		], $result->fetchAllAssociative());
+		$result->closeCursor();
+		return $rows;
+	}
+
+	private function consume(): void {
+		$this->consumer->consume(stopWhenEmpty: true);
+	}
+
+	public function testDispatchAndConsume(): void {
+		$message = new TestMessage(id: 1, text: 'hello');
+		$this->bus->dispatch($message);
+
+		$this->assertSame('high', $this->getRows()[0]['queue_name']);
+		$this->assertSame([], $this->invokableHandler->handled);
+
+		$this->consume();
+
+		$this->assertEquals([$message], $this->invokableHandler->handled);
+		$this->assertSame([], $this->getRows());
+	}
+
+	public function testDispatchRejectsMessageWithoutHandler(): void {
+		$this->expectException(InvalidMessageException::class);
+		$this->bus->dispatch(new NotAMessage(1));
+	}
+
+	public function testDispatchRejectsUnsupportedValues(): void {
+		$this->expectException(InvalidMessageException::class);
+		$this->bus->dispatch(new TestMessage(id: 1, list: [new \stdClass()]));
+	}
+
+	public function testDeduplication(): void {
+		$this->bus->dispatch(new TestMessage(id: 1, text: 'first'));
+		$this->bus->dispatch(new TestMessage(id: 1, text: 'second'));
+		$this->bus->dispatch(new TestMessage(id: 2));
+		$this->assertCount(2, $this->getRows());
+
+		$this->consume();
+		$this->assertSame(['first', 'default'], array_map(static fn (TestMessage $m): string => $m->text, $this->invokableHandler->handled));
+
+		$this->bus->dispatch(new TestMessage(id: 1, text: 'third'));
+		$this->assertCount(1, $this->getRows());
+	}
+
+	public function testUserScope(): void {
+		$this->setupManager->expects($this->once())->method('setupForUser');
+		$this->setupManager->expects($this->once())->method('tearDown');
+
+		$this->bus->dispatch(new UserMessage('alice'));
+		$this->consume();
+
+		$this->assertSame(['alice'], $this->multiHandler->users);
+		$this->assertNull($this->activeUser);
+	}
+
+	public function testUnknownUserFailsWithoutRetry(): void {
+		$this->bus->dispatch(new UserMessage('bob'));
+		$this->consume();
+
+		$this->assertSame([], $this->multiHandler->users);
+		$rows = $this->getRows();
+		$this->assertCount(1, $rows);
+		$this->assertSame('failed', $rows[0]['queue_name']);
+	}
+
+	public function testRetryThenFailureQueue(): void {
+		$this->bus->dispatch(new FailingMessage());
+		$this->consume();
+
+		$rows = $this->getRows();
+		$this->assertCount(1, $rows);
+		$this->assertSame('default', $rows[0]['queue_name']);
+		$this->assertSame(1, $rows[0]['retry_count']);
+		$this->assertSame($this->now + 30, $rows[0]['available_at']);
+
+		$this->now += 30;
+		$this->consume();
+
+		$rows = $this->getRows();
+		$this->assertCount(1, $rows);
+		$this->assertSame('failed', $rows[0]['queue_name']);
+		$this->assertSame('RuntimeException: Temporary failure', $rows[0]['last_error']);
+	}
+
+	public function testRecoverableIsRetriedWithBackoff(): void {
+		$this->bus->dispatch(new FailingMessage(recoverable: true));
+
+		foreach ([30, 60, 120] as $retry => $delay) {
+			$this->consume();
+			$rows = $this->getRows();
+			$this->assertSame('default', $rows[0]['queue_name']);
+			$this->assertSame($retry + 1, $rows[0]['retry_count']);
+			$this->assertSame($this->now + $delay, $rows[0]['available_at']);
+			$this->now += $delay;
+		}
+	}
+
+	public function testUnrecoverableGoesToFailureQueue(): void {
+		$this->bus->dispatch(new FailingMessage(unrecoverable: true));
+		$this->consume();
+
+		$rows = $this->getRows();
+		$this->assertCount(1, $rows);
+		$this->assertSame('failed', $rows[0]['queue_name']);
+		$this->assertSame(UnrecoverableMessageException::class . ': Permanent failure', $rows[0]['last_error']);
+	}
+
+	public function testUndecodableMessageIsMovedToFailureQueue(): void {
+		$qb = $this->connection->getQueryBuilder();
+		$qb->insert(MessageStore::TABLE)
+			->values([
+				'id' => $qb->createNamedParameter(Server::get(ISnowflakeGenerator::class)->nextId(), IQueryBuilder::PARAM_INT),
+				'queue_name' => $qb->createNamedParameter('default'),
+				'message_class' => $qb->createNamedParameter(NotAMessage::class),
+				'body' => $qb->createNamedParameter('{"id":1}'),
+				'created_at' => $qb->createNamedParameter($this->now, IQueryBuilder::PARAM_INT),
+				'available_at' => $qb->createNamedParameter($this->now, IQueryBuilder::PARAM_INT),
+			])
+			->executeStatement();
+
+		$this->consume();
+
+		$rows = $this->getRows();
+		$this->assertCount(1, $rows);
+		$this->assertSame('failed', $rows[0]['queue_name']);
+	}
+
+	public function testHandleAfterRequest(): void {
+		$this->session->expects($this->once())->method('close');
+		$this->lockingProvider->expects($this->once())->method('releaseAll');
+
+		$this->bus->dispatch(new TestMessage(1));
+		$this->bus->dispatch(new UserMessage('alice'));
+		$this->assertCount(1, $this->afterRequestHandler->shutdownFunctions);
+
+		$this->afterRequestHandler->run();
+
+		$this->assertTrue($this->afterRequestHandler->requestFinished);
+		$this->assertCount(1, $this->invokableHandler->handled);
+		$this->assertSame(['alice'], $this->multiHandler->users);
+		$this->assertSame([], $this->getRows());
+	}
+
+	public function testHandleAfterRequestSkipsLowMessages(): void {
+		$this->bus->dispatch(new LowMessage(1));
+		$this->assertSame([], $this->afterRequestHandler->shutdownFunctions);
+
+		$this->afterRequestHandler->run();
+
+		$this->assertFalse($this->afterRequestHandler->requestFinished);
+		$this->assertCount(1, $this->getRows());
+	}
+
+	public function testHandleAfterRequestRespectsMessageLimit(): void {
+		for ($i = 1; $i <= 7; ++$i) {
+			$this->bus->dispatch(new TestMessage($i));
+		}
+
+		$this->afterRequestHandler->run();
+
+		$this->assertSame([1, 2, 3, 4, 5], array_map(static fn (TestMessage $m): int => $m->id, $this->invokableHandler->handled));
+		$this->assertCount(2, $this->getRows());
+	}
+
+	public function testHandleAfterRequestOnlyHandlesOwnMessages(): void {
+		$this->bus->dispatch(new TestMessage(1));
+		$this->consume();
+		$this->bus->dispatch(new TestMessage(2));
+
+		$this->afterRequestHandler->run();
+
+		$this->assertSame([1, 2], array_map(static fn (TestMessage $m): int => $m->id, $this->invokableHandler->handled));
+	}
+
+	public function testHandleAfterRequestRetriesFailures(): void {
+		$this->bus->dispatch(new FailingMessage());
+
+		$this->afterRequestHandler->run();
+
+		$rows = $this->getRows();
+		$this->assertCount(1, $rows);
+		$this->assertSame('default', $rows[0]['queue_name']);
+		$this->assertSame(1, $rows[0]['retry_count']);
+	}
+}
diff --git a/version.php b/version.php
index 31e5cb2bcda4c..ca54c28709bc6 100644
--- a/version.php
+++ b/version.php
@@ -11,7 +11,7 @@
 // between betas, final and RCs. This is _not_ the public version number. Reset minor/patch level
 // when updating major/minor version number.
 
-$OC_Version = [36, 0, 0, 0];
+$OC_Version = [36, 0, 0, 1];
 
 // The human-readable string
 $OC_VersionString = '36.0.0 dev';